/*
 * Copyright 2015-2022 Transmogrify LLC, 2022-2026 Revetware LLC.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.pyranid;

import org.jspecify.annotations.NonNull;
import org.jspecify.annotations.Nullable;

import javax.annotation.concurrent.NotThreadSafe;
import javax.annotation.concurrent.ThreadSafe;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.sql.Types;
import java.time.Duration;
import java.time.ZoneId;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Deque;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Queue;
import java.util.Set;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.UnaryOperator;
import java.util.UUID;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import static java.lang.String.format;
import static java.lang.System.nanoTime;
import static java.util.Objects.requireNonNull;

/**
 * Main class for performing database access operations.
 *
 * @author <a href="https://www.revetkn.com">Mark Allen</a>
 * @since 1.0.0
 */
@ThreadSafe
public final class Database {
        @NonNull
        private static final ThreadLocal<Deque<Transaction>> TRANSACTION_STACK_HOLDER;
        private static final int DEFAULT_PARSED_SQL_CACHE_CAPACITY = 1024;
        private static final int MAX_DIAGNOSTIC_MESSAGE_LENGTH = 1024;
        private static final int MAX_DIAGNOSTIC_PARAMETER_LENGTH = 256;
        private static final int MAX_DIAGNOSTIC_PARAMETERS_LENGTH = 2048;
        private static final int MAX_DIAGNOSTIC_SQL_LENGTH = 2048;
        @NonNull
        private static final String TRUNCATED_SUFFIX = "... (truncated)";
        @NonNull
        private static final Pattern DIAGNOSTIC_WHITESPACE_PATTERN = Pattern.compile("\\s+");

        static {
                TRANSACTION_STACK_HOLDER = new ThreadLocal<>();
        }

        @NonNull
        private final DataSource dataSource;
        @NonNull
        private final AtomicReference<@Nullable DatabaseType> databaseType;
        @NonNull
        private final AtomicReference<@Nullable DatabaseDialect> databaseDialect;
        @NonNull
        private final ThreadLocal<Connection> databaseTypeDetectionConnectionHolder;
        @NonNull
        private final ZoneId timeZone;
        @NonNull
        private final AmbiguousTimestampBindingStrategy ambiguousTimestampBindingStrategy;
        @NonNull
        private final InstanceProvider instanceProvider;
        @NonNull
        private final PreparedStatementBinder preparedStatementBinder;
        @NonNull
        private final ResultSetMapper resultSetMapper;
        @NonNull
        private final StatementLogger statementLogger;
        @NonNull
        private final ParameterRedactor parameterRedactor;
        @NonNull
        private final MetricsCollectorDispatcher metricsCollectorDispatcher;
        @Nullable
        private final Duration queryTimeout;
        @Nullable
        private final Integer fetchSize;
        @Nullable
        private final Integer maxRows;
        @Nullable
        private final Map<String, ParsedSqlVariants> parsedSqlCache;
        @NonNull
        private final AtomicLong defaultIdGenerator;
        @NonNull
        private final Logger logger;

        @NonNull
        private volatile DatabaseOperationSupportStatus executeLargeBatchSupported;
        @NonNull
        private volatile DatabaseOperationSupportStatus executeLargeUpdateSupported;

        private Database(@NonNull Builder builder) {
                requireNonNull(builder);

                this.dataSource = requireNonNull(builder.dataSource);
                this.databaseType = new AtomicReference<>(builder.databaseType);
                this.databaseDialect = new AtomicReference<>(builder.databaseType == null ? null : builder.databaseType.dialect());
                this.databaseTypeDetectionConnectionHolder = new ThreadLocal<>();
                this.timeZone = builder.timeZone == null ? ZoneId.systemDefault() : builder.timeZone;
                this.ambiguousTimestampBindingStrategy = builder.ambiguousTimestampBindingStrategy == null
                                ? AmbiguousTimestampBindingStrategy.TIMESTAMP_WITH_TIME_ZONE
                                : builder.ambiguousTimestampBindingStrategy;
                this.instanceProvider = builder.instanceProvider == null ? new InstanceProvider() {} : builder.instanceProvider;
                this.preparedStatementBinder = builder.preparedStatementBinder == null ? PreparedStatementBinder.withDefaultConfiguration() : builder.preparedStatementBinder;
                this.resultSetMapper = builder.resultSetMapper == null ? ResultSetMapper.withDefaultConfiguration() : builder.resultSetMapper;
                this.statementLogger = builder.statementLogger == null ? (statementLog) -> {} : builder.statementLogger;
                this.parameterRedactor = builder.parameterRedactor == null ? ParameterRedactor.none() : builder.parameterRedactor;
                this.metricsCollectorDispatcher = new MetricsCollectorDispatcher(builder.metricsCollector);
                this.queryTimeout = validateQueryTimeout(builder.queryTimeout);
                this.fetchSize = validateNonNegativeStatementSetting("fetchSize", builder.fetchSize);
                this.maxRows = validateNonNegativeStatementSetting("maxRows", builder.maxRows);
                if (builder.parsedSqlCacheCapacity != null && builder.parsedSqlCacheCapacity < 0)
                        throw new IllegalArgumentException("parsedSqlCacheCapacity must be >= 0");

                int parsedSqlCacheCapacity = builder.parsedSqlCacheCapacity == null
                                ? DEFAULT_PARSED_SQL_CACHE_CAPACITY
                                : builder.parsedSqlCacheCapacity;
                this.parsedSqlCache = parsedSqlCacheCapacity == 0 ? null : new ConcurrentLruMap<>(parsedSqlCacheCapacity);
                this.defaultIdGenerator = new AtomicLong();
                this.logger = Logger.getLogger(getClass().getName());
                this.executeLargeBatchSupported = DatabaseOperationSupportStatus.UNKNOWN;
                this.executeLargeUpdateSupported = DatabaseOperationSupportStatus.UNKNOWN;
        }

        /**
         * Provides a {@link Database} builder for the given {@link DataSource}.
         *
         * @param dataSource data source used to create the {@link Database} builder
         * @return a {@link Database} builder
         */
        @NonNull
        public static Builder withDataSource(@NonNull DataSource dataSource) {
                requireNonNull(dataSource);
                return new Builder(dataSource);
        }

        /**
         * Gets a reference to the current transaction, if any.
         *
         * @return the current transaction
         */
        @NonNull
        public Optional<Transaction> currentTransaction() {
                @Nullable Deque<Transaction> transactionStack = TRANSACTION_STACK_HOLDER.get();
                Transaction transaction = transactionStack == null || transactionStack.isEmpty() ? null : transactionStack.peek();

                if (transaction == null || !isTransactionOwnedByThisDatabase(transaction))
                        return Optional.empty();

                return Optional.of(transaction);
        }

        @NonNull
        private Optional<Transaction> currentTransactionForDatabaseOperation() {
                @Nullable Deque<Transaction> transactionStack = TRANSACTION_STACK_HOLDER.get();
                Transaction transaction = transactionStack == null || transactionStack.isEmpty() ? null : transactionStack.peek();

                if (transaction == null)
                        return Optional.empty();

                if (!isTransactionOwnedByThisDatabase(transaction))
                        throw wrongDatabaseTransactionException(transaction);

                return Optional.of(transaction);
        }

        static boolean hasAmbientTransaction() {
                Deque<Transaction> transactionStack = TRANSACTION_STACK_HOLDER.get();
                return transactionStack != null && !transactionStack.isEmpty();
        }

        @NonNull
        private Deque<Transaction> transactionStackForPush() {
                Deque<Transaction> transactionStack = TRANSACTION_STACK_HOLDER.get();

                if (transactionStack == null) {
                        transactionStack = new ArrayDeque<>();
                        TRANSACTION_STACK_HOLDER.set(transactionStack);
                }

                return transactionStack;
        }

        private boolean isTransactionOwnedByThisDatabase(@NonNull Transaction transaction) {
                requireNonNull(transaction);
                return transaction.isOwnedBy(getDataSource());
        }

        @NonNull
        private DatabaseException wrongDatabaseTransactionException(@NonNull Transaction transaction) {
                requireNonNull(transaction);
                return new DatabaseException(format("Transaction %s belongs to a different %s than this %s. "
                                                + "Use the %s instance that created the transaction, or explicitly participate only with a transaction "
                                                + "created from the same %s.",
                                transaction.id(), DataSource.class.getSimpleName(), Database.class.getSimpleName(),
                                Database.class.getSimpleName(), DataSource.class.getSimpleName()));
        }

        /**
         * Performs an operation transactionally.
         * <p>
         * The transaction will be automatically rolled back if an exception bubbles out of {@code transactionalOperation}.
         * <p>
         * Nested calls to {@code transaction(...)} are independent transactions with independent JDBC connections; they do
         * not automatically join an outer transaction. Use {@link #participate(Transaction, TransactionalOperation)} to join an
         * existing transaction explicitly. A transaction is scoped to the {@link DataSource} instance that created it; a
         * {@link Database} using a different {@link DataSource} fails fast instead of joining it.
         *
         * @param transactionalOperation the operation to perform transactionally
         */
        public void transaction(@NonNull TransactionalOperation transactionalOperation) {
                requireNonNull(transactionalOperation);

                transaction(() -> {
                        transactionalOperation.perform();
                        return Optional.empty();
                });
        }

        /**
         * Performs an operation transactionally with the given options.
         * <p>
         * The transaction will be automatically rolled back if an exception bubbles out of {@code transactionalOperation}.
         * <p>
         * Nested calls to {@code transaction(...)} are independent transactions with independent JDBC connections; they do
         * not automatically join an outer transaction. Use {@link #participate(Transaction, TransactionalOperation)} to join an
         * existing transaction explicitly. A transaction is scoped to the {@link DataSource} instance that created it; a
         * {@link Database} using a different {@link DataSource} fails fast instead of joining it.
         *
         * @param transactionOptions     options to apply to the transaction
         * @param transactionalOperation the operation to perform transactionally
         * @since 4.2.0
         */
        public void transaction(@NonNull TransactionOptions transactionOptions,
                                                                                                        @NonNull TransactionalOperation transactionalOperation) {
                requireNonNull(transactionOptions);
                requireNonNull(transactionalOperation);

                transaction(transactionOptions, () -> {
                        transactionalOperation.perform();
                        return Optional.empty();
                });
        }

        /**
         * Performs an operation transactionally and optionally returns a value.
         * <p>
         * The transaction will be automatically rolled back if an exception bubbles out of {@code transactionalOperation}.
         * <p>
         * Nested calls to {@code transaction(...)} are independent transactions with independent JDBC connections; they do
         * not automatically join an outer transaction. Use {@link #participate(Transaction, ReturningTransactionalOperation)} to
         * join an existing transaction explicitly. A transaction is scoped to the {@link DataSource} instance that created it; a
         * {@link Database} using a different {@link DataSource} fails fast instead of joining it.
         *
         * @param transactionalOperation the operation to perform transactionally
         * @param <T>                    the type to be returned
         * @return the result of the transactional operation
         */
        @NonNull
        public <T> Optional<T> transaction(@NonNull ReturningTransactionalOperation<T> transactionalOperation) {
                requireNonNull(transactionalOperation);
                return transaction(TransactionOptions.DEFAULT, transactionalOperation);
        }

        /**
         * Performs an operation transactionally with the given options, optionally returning a value.
         * <p>
         * The transaction will be automatically rolled back if an exception bubbles out of {@code transactionalOperation}.
         * <p>
         * Nested calls to {@code transaction(...)} are independent transactions with independent JDBC connections; they do
         * not automatically join an outer transaction. Use {@link #participate(Transaction, ReturningTransactionalOperation)} to
         * join an existing transaction explicitly. A transaction is scoped to the {@link DataSource} instance that created it; a
         * {@link Database} using a different {@link DataSource} fails fast instead of joining it.
         *
         * @param transactionOptions     options to apply to the transaction
         * @param transactionalOperation the operation to perform transactionally
         * @param <T>                    the type to be returned
         * @return the result of the transactional operation
         * @since 4.2.0
         */
        @NonNull
        public <T> Optional<T> transaction(@NonNull TransactionOptions transactionOptions,
                                                                                                                                                 @NonNull ReturningTransactionalOperation<T> transactionalOperation) {
                requireNonNull(transactionOptions);
                requireNonNull(transactionalOperation);

                Transaction transaction = new Transaction(dataSource, transactionOptions, getMetricsCollectorDispatcher(),
                                peekDatabaseType(), this::getDatabaseType);
                Deque<Transaction> transactionStack = transactionStackForPush();
                transactionStack.push(transaction);
                boolean committed = false;
                boolean commitFailed = false;
                boolean rollbackFailed = false;
                boolean rollbackAttempted = false;
                Throwable thrown = null;
                long transactionStartTime = nanoTime();
                getMetricsCollectorDispatcher().didEnterTransactionClosure(transaction, transactionOptions.getIsolation(), transaction.getDatabaseType());

                try {
                        Optional<T> returnValue = transactionalOperation.perform();

                        // Safeguard in case user code accidentally returns null instead of Optional.empty()
                        if (returnValue == null)
                                returnValue = Optional.empty();

                        transaction.getConnectionLock().lock();

                        try {
                                if (transaction.isRollbackOnly()) {
                                        rollbackAttempted = true;
                                        transaction.rollback();
                                } else {
                                        try {
                                                transaction.commit();
                                        } catch (RuntimeException | Error e) {
                                                commitFailed = true;
                                                throw e;
                                        }
                                        committed = true;
                                }

                                transaction.markCompleted();
                        } finally {
                                transaction.getConnectionLock().unlock();
                        }

                        return returnValue;
                } catch (RuntimeException e) {
                        thrown = e;
                        if (rollbackAttempted) {
                                rollbackFailed = true;
                                markTransactionCompleted(transaction);
                        } else {
                                rollbackFailed = rollbackTransactionAfterFailure(transaction, e);
                        }

                        restoreInterruptIfNeeded(e);
                        throw e;
                } catch (Error e) {
                        thrown = e;
                        if (rollbackAttempted) {
                                rollbackFailed = true;
                                markTransactionCompleted(transaction);
                        } else {
                                rollbackFailed = rollbackTransactionAfterFailure(transaction, e);
                        }

                        restoreInterruptIfNeeded(e);
                        throw e;
                } catch (Throwable t) {
                        RuntimeException wrapped = new RuntimeException(t);
                        thrown = wrapped;
                        if (rollbackAttempted) {
                                rollbackFailed = true;
                                markTransactionCompleted(transaction);
                        } else {
                                rollbackFailed = rollbackTransactionAfterFailure(transaction, wrapped);
                        }

                        restoreInterruptIfNeeded(t);
                        throw wrapped;
                } finally {
                        transactionStack.pop();

                        // Ensure txn stack is fully cleaned up
                        if (transactionStack.isEmpty())
                                TRANSACTION_STACK_HOLDER.remove();

                        Throwable cleanupFailure = null;
                        boolean hadPhysicalTransaction = false;
                        boolean completionFailed = commitFailed || rollbackFailed;
                        boolean commitSerializationFailureRolledBack = commitFailed && !rollbackFailed
                                        && transaction.didCommitFailWithSerializationFailure();
                        boolean transactionOutcomeIndeterminate = completionFailed && !commitSerializationFailureRolledBack;

                        try {
                                transaction.getConnectionLock().lock();

                                try {
                                        hadPhysicalTransaction = transaction.hasConnection();

                                        if (!transaction.isCompleted())
                                                transaction.markCompleted();

                                        cleanupFailure = cleanupCompletedTransactionConnection(transaction, cleanupFailure, completionFailed);
                                } finally {
                                        transaction.getConnectionLock().unlock();
                                }
                        } finally {
                                if (transactionOutcomeIndeterminate && thrown instanceof DatabaseException)
                                        ((DatabaseException) thrown).markTransactionOutcomeIndeterminate();

                                // Execute any user-supplied post-execution hooks
                                for (Consumer<TransactionResult> postTransactionOperation : transaction.getPostTransactionOperations()) {
                                        long postTransactionStartTime = nanoTime();
                                        Throwable postTransactionThrowable = null;
                                        TransactionResult transactionResult = transactionResult(committed, transactionOutcomeIndeterminate);
                                        try {
                                                postTransactionOperation.accept(transactionResult);
                                        } catch (Throwable cleanupException) {
                                                postTransactionThrowable = cleanupException;
                                                PostTransactionOperationException postTransactionOperationException =
                                                                new PostTransactionOperationException(transactionResult, cleanupException);

                                                if (cleanupFailure == null)
                                                        cleanupFailure = postTransactionOperationException;
                                                else
                                                        cleanupFailure.addSuppressed(postTransactionOperationException);
                                        } finally {
                                                getMetricsCollectorDispatcher().didRunPostTransactionOperation(transaction, transactionResult, transaction.getDatabaseType(),
                                                                Duration.ofNanos(nanoTime() - postTransactionStartTime), postTransactionThrowable);
                                        }
                                }
                        }

                        Throwable exitThrown = thrown == null ? cleanupFailure : thrown;
                        getMetricsCollectorDispatcher().didExitTransactionClosure(transaction,
                                        transactionClosureOutcome(committed, hadPhysicalTransaction, transactionOutcomeIndeterminate),
                                        transaction.getDatabaseType(), Duration.ofNanos(nanoTime() - transactionStartTime), exitThrown);

                        if (cleanupFailure != null) {
                                if (thrown != null) {
                                        if (thrown != cleanupFailure)
                                                thrown.addSuppressed(cleanupFailure);
                                } else if (cleanupFailure instanceof RuntimeException) {
                                        if (committed && cleanupFailure instanceof DatabaseException)
                                                ((DatabaseException) cleanupFailure).markTransactionOutcomeCommitted();
                                        throw (RuntimeException) cleanupFailure;
                                } else if (cleanupFailure instanceof Error) {
                                        throw (Error) cleanupFailure;
                                } else {
                                        throw new RuntimeException(cleanupFailure);
                                }
                        }
                }
        }

        /**
         * Performs an operation transactionally, retrying according to the given retry policy.
         * <p>
         * The entire transaction closure may run more than once. Keep non-idempotent external side effects outside the closure
         * unless they are safe to repeat.
         * <p>
         * Pyranid consults the retry policy only after the database outcome is known to be rolled back. This includes a recognized
         * serialization failure reported by physical commit when the follow-up rollback succeeds. Other commit failures and all
         * rollback failures are terminal because their outcome is indeterminate.
         * <p>
         * Unlike {@link #transaction(TransactionalOperation)} and related transaction methods, retrying methods return
         * {@link TransactionRetryResult} so callers can inspect failures that were recovered before success.
         * <p>
         * This method fails fast if called inside an active transaction for this {@code Database}. Retrying a nested unit cannot
         * restart the outer transaction safely.
         *
         * @param retryPolicy           retry policy to apply
         * @param transactionalOperation the operation to perform transactionally
         * @return retry result containing any failures retried before success
         * @since 4.4.0
         */
        @NonNull
        public TransactionRetryResult<Void> transactionWithRetry(@NonNull RetryPolicy retryPolicy,
                                                                                                                                                                                                                                        @NonNull TransactionalOperation transactionalOperation) {
                requireNonNull(retryPolicy);
                requireNonNull(transactionalOperation);

                return transactionWithRetry(retryPolicy, () -> {
                        transactionalOperation.perform();
                        return Optional.<Void>empty();
                });
        }

        /**
         * Performs an operation transactionally with the given options, retrying according to the given retry policy.
         * <p>
         * The entire transaction closure may run more than once. Keep non-idempotent external side effects outside the closure
         * unless they are safe to repeat.
         * <p>
         * Pyranid consults the retry policy only after the database outcome is known to be rolled back. This includes a recognized
         * serialization failure reported by physical commit when the follow-up rollback succeeds. Other commit failures and all
         * rollback failures are terminal because their outcome is indeterminate.
         * <p>
         * Unlike {@link #transaction(TransactionOptions, TransactionalOperation)} and related transaction methods, retrying
         * methods return {@link TransactionRetryResult} so callers can inspect failures that were recovered before success.
         * <p>
         * This method fails fast if called inside an active transaction for this {@code Database}. Retrying a nested unit cannot
         * restart the outer transaction safely.
         *
         * @param retryPolicy           retry policy to apply
         * @param transactionOptions     options to apply to each transaction attempt
         * @param transactionalOperation the operation to perform transactionally
         * @return retry result containing any failures retried before success
         * @since 4.4.0
         */
        @NonNull
        public TransactionRetryResult<Void> transactionWithRetry(@NonNull RetryPolicy retryPolicy,
                                                                                                                                                                                                                                        @NonNull TransactionOptions transactionOptions,
                                                                                                                                                                                                                                        @NonNull TransactionalOperation transactionalOperation) {
                requireNonNull(retryPolicy);
                requireNonNull(transactionOptions);
                requireNonNull(transactionalOperation);

                return transactionWithRetry(retryPolicy, transactionOptions, () -> {
                        transactionalOperation.perform();
                        return Optional.<Void>empty();
                });
        }

        /**
         * Performs an operation transactionally and optionally returns a value, retrying according to the given retry policy.
         * <p>
         * The entire transaction closure may run more than once. Keep non-idempotent external side effects outside the closure
         * unless they are safe to repeat.
         * <p>
         * Pyranid consults the retry policy only after the database outcome is known to be rolled back. This includes a recognized
         * serialization failure reported by physical commit when the follow-up rollback succeeds. Other commit failures and all
         * rollback failures are terminal because their outcome is indeterminate.
         * <p>
         * Unlike {@link #transaction(ReturningTransactionalOperation)} and related transaction methods, retrying methods return
         * {@link TransactionRetryResult} so callers can inspect failures that were recovered before success.
         * <p>
         * This method fails fast if called inside an active transaction for this {@code Database}. Retrying a nested unit cannot
         * restart the outer transaction safely.
         *
         * @param retryPolicy           retry policy to apply
         * @param transactionalOperation the operation to perform transactionally
         * @param <T>                    the type to be returned
         * @return retry result containing the successful transaction value and any failures retried before success
         * @since 4.4.0
         */
        @NonNull
        public <T> TransactionRetryResult<T> transactionWithRetry(@NonNull RetryPolicy retryPolicy,
                                                                                                                                                                                                                                         @NonNull ReturningTransactionalOperation<T> transactionalOperation) {
                requireNonNull(retryPolicy);
                requireNonNull(transactionalOperation);

                return transactionWithRetry(retryPolicy, TransactionOptions.DEFAULT, transactionalOperation);
        }

        /**
         * Performs an operation transactionally with the given options and optionally returns a value, retrying according to the
         * given retry policy.
         * <p>
         * The entire transaction closure may run more than once. Keep non-idempotent external side effects outside the closure
         * unless they are safe to repeat.
         * <p>
         * Pyranid consults the retry policy only after the database outcome is known to be rolled back. This includes a recognized
         * serialization failure reported by physical commit when the follow-up rollback succeeds. Other commit failures and all
         * rollback failures are terminal because their outcome is indeterminate.
         * <p>
         * Unlike {@link #transaction(TransactionOptions, ReturningTransactionalOperation)} and related transaction methods,
         * retrying methods return {@link TransactionRetryResult} so callers can inspect failures that were recovered before
         * success.
         * <p>
         * This method fails fast if called inside an active transaction for this {@code Database}. Retrying a nested unit cannot
         * restart the outer transaction safely.
         *
         * @param retryPolicy           retry policy to apply
         * @param transactionOptions     options to apply to each transaction attempt
         * @param transactionalOperation the operation to perform transactionally
         * @param <T>                    the type to be returned
         * @return retry result containing the successful transaction value and any failures retried before success
         * @since 4.4.0
         */
        @NonNull
        public <T> TransactionRetryResult<T> transactionWithRetry(@NonNull RetryPolicy retryPolicy,
                                                                                                                                                                                                                                         @NonNull TransactionOptions transactionOptions,
                                                                                                                                                                                                                                         @NonNull ReturningTransactionalOperation<T> transactionalOperation) {
                requireNonNull(retryPolicy);
                requireNonNull(transactionOptions);
                requireNonNull(transactionalOperation);

                if (currentTransaction().isPresent())
                        throw new IllegalStateException("transactionWithRetry must not be called within an existing transaction");

                List<DatabaseException> priorFailures = new ArrayList<>();

                for (int attempt = 1; attempt <= retryPolicy.getMaxAttempts(); ++attempt) {
                        try {
                                return TransactionRetryResult.of(transaction(transactionOptions, transactionalOperation), priorFailures);
                        } catch (DatabaseException e) {
                                if (e.isTransactionOutcomeRetryUnsafe()) {
                                        suppressPriorFailures(e, priorFailures);
                                        throw e;
                                }

                                boolean finalAttempt = attempt == retryPolicy.getMaxAttempts();

                                if (finalAttempt) {
                                        suppressPriorFailures(e, priorFailures);
                                        throw e;
                                }

                                Boolean retryable;

                                try {
                                        retryable = retryPolicy.getCondition().shouldRetry(e);
                                } catch (RuntimeException | Error conditionFailure) {
                                        suppressRetryFailures(conditionFailure, e, priorFailures);
                                        throw conditionFailure;
                                }

                                if (retryable == null) {
                                        NullPointerException nullConditionFailure =
                                                        new NullPointerException("RetryPolicy.Condition returned null");
                                        suppressRetryFailures(nullConditionFailure, e, priorFailures);
                                        throw nullConditionFailure;
                                }

                                if (!retryable) {
                                        suppressPriorFailures(e, priorFailures);
                                        throw e;
                                }

                                priorFailures.add(e);

                                Duration delay;

                                try {
                                        delay = retryPolicy.getBackoff().delayAfterFailedAttempt(attempt, e);
                                } catch (RuntimeException | Error backoffFailure) {
                                        suppressPriorFailures(backoffFailure, priorFailures);
                                        throw backoffFailure;
                                }

                                if (delay == null) {
                                        NullPointerException nullBackoffFailure = new NullPointerException("RetryPolicy.Backoff returned null");
                                        suppressPriorFailures(nullBackoffFailure, priorFailures);
                                        throw nullBackoffFailure;
                                }

                                if (delay.isNegative()) {
                                        IllegalArgumentException negativeBackoffFailure =
                                                        new IllegalArgumentException("RetryPolicy.Backoff returned a negative delay");
                                        suppressPriorFailures(negativeBackoffFailure, priorFailures);
                                        throw negativeBackoffFailure;
                                }

                                try {
                                        sleepBackoff(delay);
                                } catch (InterruptedException interruptedException) {
                                        Thread.currentThread().interrupt();
                                        suppressPriorFailures(e, priorFailures);
                                        e.addSuppressed(interruptedException);
                                        throw e;
                                }
                        }
                }

                throw new AssertionError("unreachable");
        }

        private void sleepBackoff(@NonNull Duration delay) throws InterruptedException {
                requireNonNull(delay);

                if (delay.isZero())
                        return;

                long millis;
                int nanos;

                try {
                        millis = delay.toMillis();
                        Duration remainder = delay.minusMillis(millis);
                        nanos = (int) Math.min(999_999L, Math.max(0L, remainder.toNanos()));
                } catch (ArithmeticException e) {
                        millis = Long.MAX_VALUE;
                        nanos = 999_999;
                }

                Thread.sleep(millis, nanos);
        }

        private void suppressRetryFailures(@NonNull Throwable failure,
                                                                                                                                                 @NonNull DatabaseException currentFailure,
                                                                                                                                                 @NonNull List<@NonNull DatabaseException> priorFailures) {
                requireNonNull(failure);
                requireNonNull(currentFailure);
                requireNonNull(priorFailures);

                suppressPriorFailures(failure, priorFailures);
                suppressIfDifferent(failure, currentFailure);
        }

        private void suppressPriorFailures(@NonNull Throwable failure,
                                                                                                                                                 @NonNull List<@NonNull DatabaseException> priorFailures) {
                requireNonNull(failure);
                requireNonNull(priorFailures);

                for (DatabaseException priorFailure : priorFailures)
                        suppressIfDifferent(failure, priorFailure);
        }

        private void suppressIfDifferent(@NonNull Throwable failure,
                                                                                                                                         @NonNull Throwable suppressed) {
                requireNonNull(failure);
                requireNonNull(suppressed);

                if (failure != suppressed)
                        failure.addSuppressed(suppressed);
        }

        private boolean rollbackTransactionAfterFailure(@NonNull Transaction transaction,
                                                                                                                                                                                                 @NonNull Throwable primary) {
                requireNonNull(transaction);
                requireNonNull(primary);

                boolean rollbackFailed = false;
                transaction.getConnectionLock().lock();

                try {
                        try {
                                transaction.rollback();
                        } catch (Throwable rollbackException) {
                                rollbackFailed = true;
                                if (primary != rollbackException)
                                        primary.addSuppressed(rollbackException);
                        } finally {
                                transaction.markCompleted();
                        }
                } finally {
                        transaction.getConnectionLock().unlock();
                }

                return rollbackFailed;
        }

        private void markTransactionCompleted(@NonNull Transaction transaction) {
                requireNonNull(transaction);
                transaction.getConnectionLock().lock();

                try {
                        transaction.markCompleted();
                } finally {
                        transaction.getConnectionLock().unlock();
                }
        }

        @Nullable
        private Throwable cleanupCompletedTransactionConnection(@NonNull Transaction transaction,
                                                                                                                                                                                         @Nullable Throwable cleanupFailure,
                                                                                                                                                                                         boolean completionFailed) {
                requireNonNull(transaction);
                boolean abortRequired = completionFailed;

                // A failed commit or rollback makes the connection unsafe to restore and reuse, even when a recognized serialization
                // failure plus successful rollback proves the logical outcome. When the outcome is unknown, restoring auto-commit to
                // true is especially dangerous because it can commit work that Pyranid meant to roll back.
                if (!abortRequired) {
                        try {
                                transaction.restoreTransactionIsolationIfNeeded();
                        } catch (Throwable cleanupException) {
                                cleanupFailure = appendSuppressed(cleanupFailure, cleanupException);
                                abortRequired = true;
                        }

                        if (!abortRequired) {
                                try {
                                        transaction.restoreReadOnlyIfNeeded();
                                } catch (Throwable cleanupException) {
                                        cleanupFailure = appendSuppressed(cleanupFailure, cleanupException);
                                        abortRequired = true;
                                }
                        }

                        if (!abortRequired && transaction.getInitialAutoCommit().isPresent() && transaction.getInitialAutoCommit().get()) {
                                try {
                                        // Autocommit was true initially, so restoring to true now that transaction has completed
                                        transaction.setAutoCommit(true);
                                } catch (Throwable cleanupException) {
                                        cleanupFailure = appendSuppressed(cleanupFailure, cleanupException);
                                        abortRequired = true;
                                }
                        }
                }

                Connection connection = transaction.getExistingConnection().orElse(null);

                if (connection != null) {
                        Duration heldDuration = transaction.getConnectionAcquiredAtNanos()
                                        .map(acquiredAtNanos -> Duration.ofNanos(nanoTime() - acquiredAtNanos))
                                        .orElse(Duration.ZERO);

                        if (abortRequired) {
                                try {
                                        abortConnection(connection);
                                } catch (Throwable cleanupException) {
                                        cleanupFailure = appendSuppressed(cleanupFailure, cleanupException);
                                }
                        }

                        try {
                                closeConnection(connection);
                                getMetricsCollectorDispatcher().didReleaseTransactionConnection(transaction, transaction.getDatabaseType(), heldDuration);
                        } catch (Throwable cleanupException) {
                                getMetricsCollectorDispatcher().didFailToReleaseTransactionConnection(transaction, transaction.getDatabaseType(), heldDuration, cleanupException);
                                cleanupFailure = appendSuppressed(cleanupFailure, cleanupException);
                        } finally {
                                // A completed Transaction must never retain a discarded or closed connection handle, including when close fails.
                                transaction.clearConnection();
                        }
                }

                return cleanupFailure;
        }

        @NonNull
        private static Throwable appendSuppressed(@Nullable Throwable existing,
                                                                                                                                                                                @NonNull Throwable additional) {
                requireNonNull(additional);

                if (existing == null)
                        return additional;

                if (existing != additional)
                        existing.addSuppressed(additional);

                return existing;
        }

        private void closeConnection(@NonNull Connection connection) {
                requireNonNull(connection);

                try {
                        connection.close();
                } catch (SQLException e) {
                        throw new DatabaseException("Unable to close database connection", e);
                }
        }

        private void abortConnection(@NonNull Connection connection) {
                requireNonNull(connection);

                try {
                        connection.abort(Runnable::run);
                } catch (SQLException e) {
                        throw new DatabaseException("Unable to abort database connection", e);
                }
        }

        @NonNull
        private static DatabaseException databaseExceptionWithStatementContext(@NonNull StatementContext<?> statementContext,
                                                                                                                                                                                                                                                        @NonNull Throwable cause) {
                requireNonNull(statementContext);
                requireNonNull(cause);

                if (cause instanceof DatabaseException databaseException && databaseException.areStatementDiagnosticsApplied())
                        return databaseException;

                String message = cause.getMessage();

                if (message == null || message.trim().length() == 0)
                        message = "Database operation failed";

                if (cause instanceof DatabaseException databaseException) {
                        SecureParameterSupport.DiagnosticScrub diagnosticScrub = diagnosticScrubForStatementContext(statementContext);

                        if (diagnosticScrub.redactor() == null && diagnosticScrub.needleRenderingFailureDescriptions().isEmpty())
                                return databaseException;

                        return databaseExceptionWithStatementContext(statementContext, message, cause, diagnosticScrub);
                }

                return databaseExceptionWithStatementContext(statementContext, message, cause);
        }

        @NonNull
        private static DatabaseException databaseExceptionWithStatementContext(@NonNull StatementContext<?> statementContext,
                                                                                                                                                                                                                                                                                                @NonNull String message,
                                                                                                                                                                                                                                                                                                @NonNull Throwable cause) {
                requireNonNull(statementContext);
                requireNonNull(message);
                requireNonNull(cause);

                SecureParameterSupport.DiagnosticScrub diagnosticScrub = diagnosticScrubForStatementContext(statementContext);
                return databaseExceptionWithStatementContext(statementContext, message, cause, diagnosticScrub);
        }

        @NonNull
        private static DatabaseException databaseExceptionWithStatementContext(@NonNull StatementContext<?> statementContext,
                                                                                                                                                                                                                                                        @NonNull String message,
                                                                                                                                                                                                                                                        @NonNull Throwable cause,
                                                                                                                                                                                                                                                        SecureParameterSupport.@NonNull DiagnosticScrub diagnosticScrub) {
                requireNonNull(statementContext);
                requireNonNull(message);
                requireNonNull(cause);
                requireNonNull(diagnosticScrub);

                UnaryOperator<String> diagnosticRedactor = diagnosticScrub.redactor();

                // Scrub the RAW message before whitespace-collapsing/truncation: bounding first would break needles
                // containing consecutive whitespace and can cut a secret mid-way, leaving an unmatchable prefix
                String scrubbedMessage = diagnosticRedactor == null ? message : diagnosticRedactor.apply(message);

                StatementDiagnostic statementDiagnostic = statementDiagnostic(statementContext);
                String diagnosticMessage = format("%s [%s]",
                                boundedDiagnosticMessage(scrubbedMessage), statementDiagnostic.diagnostic());
                DatabaseException databaseException = cause instanceof DatabaseException original
                                ? new DatabaseException(diagnosticMessage, original, diagnosticRedactor)
                                : new DatabaseException(diagnosticMessage, cause,
                                                databaseDialectForException(statementContext, cause), diagnosticRedactor);
                databaseException.markStatementDiagnosticsApplied();

                if (statementDiagnostic.parameterRenderingFailure() != null)
                        databaseException.addSuppressed(statementDiagnostic.parameterRenderingFailure());

                for (String failureDescription : diagnosticScrub.needleRenderingFailureDescriptions())
                        databaseException.addSuppressed(new RuntimeException(format(
                                        "Unable to render a secure parameter for diagnostic scrubbing: %s", failureDescription)));

                return databaseException;
        }

        private static SecureParameterSupport.@NonNull DiagnosticScrub diagnosticScrubForStatementContext(@NonNull StatementContext<?> statementContext) {
                requireNonNull(statementContext);

                try {
                        return SecureParameterSupport.diagnosticScrubForParameters(statementContext.getParameters());
                } catch (Throwable ignored) {
                        // Per-needle failures are handled inside diagnosticScrubForParameters; reaching here would be
                        // catastrophic (a plain getter throwing) - fall back to no scrubbing rather than losing the exception
                        return SecureParameterSupport.DiagnosticScrub.NONE;
                }
        }

        @NonNull
        private static DatabaseDialect databaseDialectForException(@NonNull StatementContext<?> statementContext,
                                                                                                                                                                                                                                                 @NonNull Throwable cause) {
                requireNonNull(statementContext);
                requireNonNull(cause);

                try {
                        return statementContext.getDatabaseDialect();
                } catch (Throwable ignored) {
                        return DatabaseDialect.forExceptionCause(cause);
                }
        }

        @NonNull
        private DatabaseException databaseExceptionWithRawConnectionContext(@Nullable Connection connection,
                                                                                                                                                                                                                                                                                        @NonNull Exception cause) {
                requireNonNull(cause);

                return new DatabaseException(cause.getMessage(), cause, databaseDialectForRawConnectionException(connection, cause));
        }

        @NonNull
        private DatabaseDialect databaseDialectForRawConnectionException(@Nullable Connection connection,
                                                                                                                                                                                                                                                                        @NonNull Throwable cause) {
                requireNonNull(cause);

                if (connection != null) {
                        try {
                                return getDatabaseDialect(connection);
                        } catch (Throwable ignored) {
                                // Fall through to cause-based classification below.
                        }
                }

                return DatabaseDialect.forExceptionCause(cause);
        }

        @NonNull
        private static StatementDiagnostic statementDiagnostic(@NonNull StatementContext<?> statementContext) {
                requireNonNull(statementContext);

                Statement statement = statementContext.getStatement();
                String parameters;
                Throwable parameterRenderingFailure = null;

                try {
                        parameters = boundedDiagnosticParameters(statementContext.getRedactedParameters());
                } catch (Throwable t) {
                        parameters = "<unavailable>";
                        parameterRenderingFailure = t;
                }

                String diagnostic = format("statementId=%s, sql=%s, parameters=%s",
                                statement.getId(), boundedSql(statement.getSql()), parameters);

                if (parameterRenderingFailure != null)
                        diagnostic = format("%s, parameterRenderingFailure=%s", diagnostic,
                                        boundedDiagnosticParameter(parameterRenderingFailureDiagnostic(parameterRenderingFailure)));

                return new StatementDiagnostic(diagnostic, parameterRenderingFailure);
        }

        private record StatementDiagnostic(@NonNull String diagnostic,
                                                                                                                                                 @Nullable Throwable parameterRenderingFailure) {}

        @NonNull
        private static String parameterRenderingFailureDiagnostic(@NonNull Throwable parameterRenderingFailure) {
                requireNonNull(parameterRenderingFailure);

                String message;

                try {
                        message = parameterRenderingFailure.getMessage();
                } catch (Throwable ignored) {
                        message = null;
                }

                if (message == null || message.trim().length() == 0)
                        return parameterRenderingFailure.getClass().getName();

                return format("%s: %s", parameterRenderingFailure.getClass().getName(), message);
        }

        @NonNull
        private static TransactionResult transactionResult(@NonNull Boolean committed,
                                                                                                                                                                 @NonNull Boolean transactionOutcomeIndeterminate) {
                requireNonNull(committed);
                requireNonNull(transactionOutcomeIndeterminate);

                if (committed)
                        return TransactionResult.COMMITTED;

                return transactionOutcomeIndeterminate ? TransactionResult.IN_DOUBT : TransactionResult.ROLLED_BACK;
        }

        private static MetricsCollector.TransactionClosureOutcome transactionClosureOutcome(@NonNull Boolean committed,
                                                                                                                                                                                                                                         @NonNull Boolean hadPhysicalTransaction,
                                                                                                                                                                                                                                         @NonNull Boolean transactionOutcomeIndeterminate) {
                requireNonNull(committed);
                requireNonNull(hadPhysicalTransaction);
                requireNonNull(transactionOutcomeIndeterminate);

                if (!hadPhysicalTransaction)
                        return MetricsCollector.TransactionClosureOutcome.NO_PHYSICAL_TX;

                if (committed)
                        return MetricsCollector.TransactionClosureOutcome.COMMITTED;

                return transactionOutcomeIndeterminate
                                ? MetricsCollector.TransactionClosureOutcome.FAILED
                                : MetricsCollector.TransactionClosureOutcome.ROLLED_BACK;
        }

        @Nullable
        static Long sumBatchUpdateCounts(@NonNull List<Long> updateCounts) {
                requireNonNull(updateCounts);

                long total = 0L;

                for (Long updateCount : updateCounts) {
                        if (updateCount == null || updateCount < 0L)
                                return null;

                        try {
                                total = Math.addExact(total, updateCount);
                        } catch (ArithmeticException e) {
                                return null;
                        }
                }

                return total;
        }

        @NonNull
        private static String boundedDiagnosticMessage(@NonNull String message) {
                requireNonNull(message);

                String compactMessage = DIAGNOSTIC_WHITESPACE_PATTERN.matcher(message).replaceAll(" ").trim();

                if (compactMessage.length() <= MAX_DIAGNOSTIC_MESSAGE_LENGTH)
                        return compactMessage;

                int prefixLength = Math.max(0, MAX_DIAGNOSTIC_MESSAGE_LENGTH - TRUNCATED_SUFFIX.length());
                return compactMessage.substring(0, prefixLength) + TRUNCATED_SUFFIX;
        }

        @NonNull
        private static String boundedSql(@NonNull String sql) {
                requireNonNull(sql);

                String compactSql = DIAGNOSTIC_WHITESPACE_PATTERN.matcher(sql).replaceAll(" ").trim();

                if (compactSql.length() <= MAX_DIAGNOSTIC_SQL_LENGTH)
                        return compactSql;

                int prefixLength = Math.max(0, MAX_DIAGNOSTIC_SQL_LENGTH - TRUNCATED_SUFFIX.length());
                return compactSql.substring(0, prefixLength) + TRUNCATED_SUFFIX;
        }

        @NonNull
        private static String boundedDiagnosticParameters(@NonNull List<@Nullable Object> parameters) {
                requireNonNull(parameters);

                StringBuilder parametersBuilder = new StringBuilder("[");

                for (int i = 0; i < parameters.size(); ++i) {
                        String separator = i == 0 ? "" : ", ";
                        String renderedParameter = boundedDiagnosticParameter(parameters.get(i));
                        int requiredLength = parametersBuilder.length() + separator.length() + renderedParameter.length() + 1;

                        if (requiredLength > MAX_DIAGNOSTIC_PARAMETERS_LENGTH) {
                                int availableLength = MAX_DIAGNOSTIC_PARAMETERS_LENGTH
                                                - parametersBuilder.length()
                                                - separator.length()
                                                - TRUNCATED_SUFFIX.length()
                                                - 1;

                                parametersBuilder.append(separator);

                                if (availableLength > 0)
                                        parametersBuilder.append(renderedParameter, 0, Math.min(availableLength, renderedParameter.length()));

                                parametersBuilder.append(TRUNCATED_SUFFIX).append(']');
                                if (parametersBuilder.length() > MAX_DIAGNOSTIC_PARAMETERS_LENGTH) {
                                        parametersBuilder.setLength(MAX_DIAGNOSTIC_PARAMETERS_LENGTH - 1);
                                        parametersBuilder.append(']');
                                }
                                return parametersBuilder.toString();
                        }

                        parametersBuilder.append(separator).append(renderedParameter);
                }

                parametersBuilder.append(']');
                return parametersBuilder.toString();
        }

        @NonNull
        private static String boundedDiagnosticParameter(@Nullable Object parameter) {
                String renderedParameter = String.valueOf(parameter);
                String compactParameter = DIAGNOSTIC_WHITESPACE_PATTERN.matcher(renderedParameter).replaceAll(" ").trim();

                if (compactParameter.length() <= MAX_DIAGNOSTIC_PARAMETER_LENGTH)
                        return compactParameter;

                int prefixLength = Math.max(0, MAX_DIAGNOSTIC_PARAMETER_LENGTH - TRUNCATED_SUFFIX.length());
                return compactParameter.substring(0, prefixLength) + TRUNCATED_SUFFIX;
        }

        /**
         * Performs an operation in the context of a pre-existing transaction.
         * <p>
         * No commit or rollback on the transaction will occur when {@code transactionalOperation} completes.
         * <p>
         * However, if an exception bubbles out of {@code transactionalOperation}, the transaction will be marked as rollback-only.
         * <p>
         * The transaction must have been created by this {@link Database}, or by another {@link Database} using the same
         * {@link DataSource} instance.
         * <p>
         * If this thread is interrupted while waiting for another participant to release the transaction connection, Pyranid
         * restores the interrupt flag and throws {@link DatabaseException}.
         *
         * @param transaction            the transaction in which to participate
         * @param transactionalOperation the operation that should participate in the transaction
         */
        public void participate(@NonNull Transaction transaction,
                                                                                                        @NonNull TransactionalOperation transactionalOperation) {
                requireNonNull(transaction);
                requireNonNull(transactionalOperation);

                participate(transaction, () -> {
                        transactionalOperation.perform();
                        return Optional.empty();
                });
        }

        /**
         * Performs an operation in the context of a pre-existing transaction, optionally returning a value.
         * <p>
         * No commit or rollback on the transaction will occur when {@code transactionalOperation} completes.
         * <p>
         * However, if an exception bubbles out of {@code transactionalOperation}, the transaction will be marked as rollback-only.
         * <p>
         * The transaction must have been created by this {@link Database}, or by another {@link Database} using the same
         * {@link DataSource} instance.
         * <p>
         * If this thread is interrupted while waiting for another participant to release the transaction connection, Pyranid
         * restores the interrupt flag and throws {@link DatabaseException}.
         *
         * @param transaction            the transaction in which to participate
         * @param transactionalOperation the operation that should participate in the transaction
         * @param <T>                    the type to be returned
         * @return the result of the transactional operation
         */
        @NonNull
        public <T> Optional<T> participate(@NonNull Transaction transaction,
                                                                                                                                                 @NonNull ReturningTransactionalOperation<T> transactionalOperation) {
                requireNonNull(transaction);
                requireNonNull(transactionalOperation);

                if (!isTransactionOwnedByThisDatabase(transaction))
                        throw wrongDatabaseTransactionException(transaction);

                if (transaction.isCompleted())
                        throw new IllegalStateException(format("Transaction %s has already completed and cannot participate", transaction.id()));

                Deque<Transaction> transactionStack = transactionStackForPush();
                transactionStack.push(transaction);

                try {
                        Optional<T> returnValue = transactionalOperation.perform();
                        return returnValue == null ? Optional.empty() : returnValue;
                } catch (RuntimeException e) {
                        setRollbackOnlyAfterParticipationFailure(transaction, e);
                        restoreInterruptIfNeeded(e);
                        throw e;
                } catch (Error e) {
                        setRollbackOnlyAfterParticipationFailure(transaction, e);
                        restoreInterruptIfNeeded(e);
                        throw e;
                } catch (Throwable t) {
                        RuntimeException wrapped = new RuntimeException(t);
                        setRollbackOnlyAfterParticipationFailure(transaction, wrapped);
                        restoreInterruptIfNeeded(t);
                        throw wrapped;
                } finally {
                        try {
                                transactionStack.pop();
                        } finally {
                                if (transactionStack.isEmpty())
                                        TRANSACTION_STACK_HOLDER.remove();
                        }
                }
        }

        private void setRollbackOnlyAfterParticipationFailure(@NonNull Transaction transaction,
                                                                                                                                                                                                                                @NonNull Throwable primary) {
                requireNonNull(transaction);
                requireNonNull(primary);

                try {
                        transaction.setRollbackOnly(true);
                } catch (IllegalStateException e) {
                        primary.addSuppressed(e);
                }
        }

        /**
         * Creates a fluent builder for executing SQL.
         * <p>
         * Named parameters use the {@code :paramName} syntax and are bound via {@link Query#bind(String, Object)}.
         * Positional parameters via {@code ?} are not supported.
         * Pyranid ignores parameter-looking text inside SQL string literals, quoted identifiers, comments, PostgreSQL
         * dollar-quoted strings, and SQL Server-style bracket-quoted identifiers. PostgreSQL JSONB/hstore {@code ?},
         * {@code ?|}, and {@code ?&} operators are supported; when running against {@link DatabaseType#POSTGRESQL}, Pyranid
         * emits pgjdbc's escaped {@code ??} form automatically. Unterminated quotes and comments fail fast.
         * <p>
         * Example:
         * <pre>{@code
         * Optional<Employee> employee = database.query("SELECT * FROM employee WHERE id = :id")
         *   .bind("id", 42)
         *   .fetchObject(Employee.class);
         * }</pre>
         *
         * @param sql SQL containing {@code :paramName} placeholders
         * @return a fluent builder for binding parameters and executing
         * @since 4.0.0
         */
        @NonNull
        public Query query(@NonNull String sql) {
                requireNonNull(sql);
                return new DefaultQuery(this, sql);
        }

        /**
         * Performs a portable connectivity check using JDBC {@link Connection#isValid(int)}.
         * <p>
         * This method borrows a fresh connection from this database's {@link DataSource}, calls
         * {@link Connection#isValid(int)}, and closes the connection before returning. It does <strong>not</strong>
         * participate in an active Pyranid transaction, if one exists.
         * <p>
         * JDBC accepts timeout values in whole seconds. Positive sub-second durations are rounded up to one second;
         * {@link Duration#ZERO} passes a timeout of {@code 0} to the driver.
         *
         * @param timeout maximum time to wait for driver validation
         * @throws IllegalArgumentException if {@code timeout} is negative or too large for JDBC's integer-second timeout
         * @throws DatabaseException if connection acquisition fails, validation throws, or the driver reports the
         *                           connection is not valid
         * @since 4.2.0
         */
        public void performHealthCheck(@NonNull Duration timeout) {
                requireNonNull(timeout);
                int timeoutSeconds = healthCheckTimeoutSeconds(timeout);

                performRawConnectionOperation(connection -> {
                        boolean valid;

                        try {
                                valid = connection.isValid(timeoutSeconds);
                        } catch (SQLException e) {
                                throw new DatabaseException("Unable to perform database health check", e);
                        }

                        if (!valid)
                                throw new DatabaseException("Database health check failed: connection is not valid");

                        return Optional.empty();
                }, false);
        }

        private static void restoreInterruptIfNeeded(@NonNull Throwable throwable) {
                requireNonNull(throwable);

                Throwable current = throwable;

                while (current != null) {
                        if (current instanceof InterruptedException) {
                                Thread.currentThread().interrupt();
                                return;
                        }

                        current = current.getCause();
                }
        }

        private static void lockInterruptibly(@NonNull ReentrantLock lock,
                                                                                                                                                                @NonNull String operation) {
                requireNonNull(lock);
                requireNonNull(operation);

                try {
                        lock.lockInterruptibly();
                } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        throw new DatabaseException(format("Interrupted while waiting to %s", operation), e);
                }
        }

        @Nullable
        private static Throwable closeStatementContextResources(@NonNull StatementContext<?> statementContext,
                                                                                                                                                                                                                                        @Nullable Throwable cleanupFailure) {
                requireNonNull(statementContext);

                Queue<AutoCloseable> cleanupOperations = statementContext.getCleanupOperations();
                AutoCloseable cleanupOperation;

                while ((cleanupOperation = cleanupOperations.poll()) != null) {
                        try {
                                cleanupOperation.close();
                        } catch (Throwable cleanupException) {
                                if (cleanupFailure == null)
                                        cleanupFailure = cleanupException;
                                else
                                        cleanupFailure.addSuppressed(cleanupException);
                        }
                }

                return cleanupFailure;
        }

        private static boolean isUnsupportedSqlFeature(@NonNull SQLException e) {
                requireNonNull(e);

                String sqlState = e.getSQLState();
                if (sqlState != null) {
                        if (sqlState.startsWith("0A") || "HYC00".equals(sqlState))
                                return true;
                }

                Throwable cause = e.getCause();
                if (cause instanceof SQLFeatureNotSupportedException
                                || cause instanceof UnsupportedOperationException
                                || cause instanceof AbstractMethodError)
                        return true;

                String message = e.getMessage();
                if (message == null)
                        return false;

                String lower = message.toLowerCase(Locale.ROOT);
                return lower.contains("not supported")
                                || lower.contains("unsupported")
                                || lower.contains("not implemented")
                                || lower.contains("feature not supported");
        }

        @Nullable
        private static Duration validateQueryTimeout(@Nullable Duration queryTimeout) {
                if (queryTimeout != null) {
                        if (queryTimeout.isNegative())
                                throw new IllegalArgumentException("queryTimeout must be >= 0");

                        queryTimeoutSeconds(queryTimeout);
                }

                return queryTimeout;
        }

        @Nullable
        private static Integer validateNonNegativeStatementSetting(@NonNull String name,
                                                                                                                                                                                                                                                 @Nullable Integer value) {
                requireNonNull(name);

                if (value != null && value < 0)
                        throw new IllegalArgumentException(format("%s must be >= 0", name));

                return value;
        }

        @Nullable
        private static Integer validatePositiveQuerySetting(@NonNull String name,
                                                                                                                                                                                                                 @Nullable Integer value) {
                requireNonNull(name);

                if (value != null && value <= 0)
                        throw new IllegalArgumentException(format("%s must be > 0", name));

                return value;
        }

        private static int queryTimeoutSeconds(@NonNull Duration queryTimeout) {
                requireNonNull(queryTimeout);

                long seconds = queryTimeout.getSeconds();

                if (queryTimeout.getNano() > 0) {
                        if (seconds == Long.MAX_VALUE)
                                throw new IllegalArgumentException(format("queryTimeout must be <= %s seconds", Integer.MAX_VALUE));

                        ++seconds;
                }

                if (seconds > Integer.MAX_VALUE)
                        throw new IllegalArgumentException(format("queryTimeout must be <= %s seconds", Integer.MAX_VALUE));

                return (int) seconds;
        }

        private static int healthCheckTimeoutSeconds(@NonNull Duration timeout) {
                requireNonNull(timeout);

                if (timeout.isNegative())
                        throw new IllegalArgumentException("timeout must be >= 0");

                long seconds = timeout.getSeconds();

                if (timeout.getNano() > 0) {
                        if (seconds == Long.MAX_VALUE)
                                throw new IllegalArgumentException(format("timeout must be <= %s seconds", Integer.MAX_VALUE));

                        ++seconds;
                }

                if (seconds > Integer.MAX_VALUE)
                        throw new IllegalArgumentException(format("timeout must be <= %s seconds", Integer.MAX_VALUE));

                return (int) seconds;
        }

        @NonNull
        private ParsedSqlVariants getParsedSqlVariants(@NonNull String sql) {
                requireNonNull(sql);

                if (this.parsedSqlCache == null)
                        return parseNamedParameterSqlVariants(sql);

                return this.parsedSqlCache.computeIfAbsent(sql, Database::parseNamedParameterSqlVariants);
        }

        @NonNull
        private static ParsedSqlVariants parseNamedParameterSqlVariants(@NonNull String sql) {
                requireNonNull(sql);

                ParsedSql standardParsedSql = null;
                ParsedSql mysqlParsedSql = null;
                IllegalArgumentException standardParseFailure = null;
                IllegalArgumentException mysqlParseFailure = null;

                try {
                        standardParsedSql = parseNamedParameterSql(sql, SqlLexicalMode.STANDARD);
                } catch (IllegalArgumentException e) {
                        standardParseFailure = e;
                }

                if (requiresMySqlLexicalVariant(sql)) {
                        try {
                                mysqlParsedSql = parseNamedParameterSql(sql, SqlLexicalMode.MYSQL);
                        } catch (IllegalArgumentException e) {
                                mysqlParseFailure = e;
                        }
                } else {
                        mysqlParsedSql = standardParsedSql;
                        mysqlParseFailure = standardParseFailure;
                }

                if (standardParsedSql == null && mysqlParsedSql == null)
                        throw requireNonNull(standardParseFailure);

                ParsedSql effectiveStandardParsedSql = standardParsedSql == null
                                ? requireNonNull(mysqlParsedSql)
                                : standardParsedSql;
                ParsedSql effectiveMysqlParsedSql = mysqlParsedSql == null
                                ? effectiveStandardParsedSql
                                : mysqlParsedSql;
                Set<String> distinctParameterNames;

                if (effectiveStandardParsedSql.distinctParameterNames.equals(effectiveMysqlParsedSql.distinctParameterNames)) {
                        distinctParameterNames = effectiveStandardParsedSql.distinctParameterNames;
                } else {
                        Set<String> combinedParameterNames = new HashSet<>(effectiveStandardParsedSql.distinctParameterNames);
                        combinedParameterNames.addAll(effectiveMysqlParsedSql.distinctParameterNames);
                        distinctParameterNames = Set.copyOf(combinedParameterNames);
                }

                boolean requiresDatabaseType = standardParseFailure != null
                                || mysqlParseFailure != null
                                || !effectiveStandardParsedSql.equivalentTo(effectiveMysqlParsedSql);

                return new ParsedSqlVariants(effectiveStandardParsedSql,
                                effectiveMysqlParsedSql,
                                standardParseFailure == null ? null : requireNonNull(standardParseFailure.getMessage()),
                                mysqlParseFailure == null ? null : requireNonNull(mysqlParseFailure.getMessage()),
                                distinctParameterNames,
                                requiresDatabaseType);
        }

        private static boolean requiresMySqlLexicalVariant(@NonNull String sql) {
                requireNonNull(sql);
                return sql.indexOf('#') >= 0 || sql.indexOf("\\'") >= 0;
        }

        private static final class ParsedSqlVariants {
                @NonNull
                private final ParsedSql standardParsedSql;
                @NonNull
                private final ParsedSql mysqlParsedSql;
                @Nullable
                private final String standardParseFailureMessage;
                @Nullable
                private final String mysqlParseFailureMessage;
                @NonNull
                private final Set<String> distinctParameterNames;
                private final boolean requiresDatabaseType;

                private ParsedSqlVariants(@NonNull ParsedSql standardParsedSql,
                                                                                                                @NonNull ParsedSql mysqlParsedSql,
                                                                                                                @Nullable String standardParseFailureMessage,
                                                                                                                @Nullable String mysqlParseFailureMessage,
                                                                                                                @NonNull Set<String> distinctParameterNames,
                                                                                                                boolean requiresDatabaseType) {
                        this.standardParsedSql = requireNonNull(standardParsedSql);
                        this.mysqlParsedSql = requireNonNull(mysqlParsedSql);
                        this.standardParseFailureMessage = standardParseFailureMessage;
                        this.mysqlParseFailureMessage = mysqlParseFailureMessage;
                        this.distinctParameterNames = Set.copyOf(requireNonNull(distinctParameterNames));
                        this.requiresDatabaseType = requiresDatabaseType;
                }
        }

        /**
         * Default internal implementation of {@link Query}.
         * <p>
         * This class is intended for use by a single thread.
         */
        @NotThreadSafe
        private static final class DefaultQuery implements Query {
                @NonNull
                private final Database database;
                @NonNull
                private final String originalSql;
                @NonNull
                private final ParsedSqlVariants parsedSqlVariants;
                @NonNull
                private final Map<String, Object> bindings;
                @Nullable
                private PreparedStatementCustomizer preparedStatementCustomizer;
                @Nullable
                private Duration queryTimeout;
                @Nullable
                private Integer fetchSize;
                @Nullable
                private Integer maxRows;
                @Nullable
                private Integer batchChunkSize;
                @Nullable
                private Object id;
                @Nullable
                private ResultSetMapper resultSetMapper;
                @Nullable
                private PreparedStatementBinder preparedStatementBinder;

                private DefaultQuery(@NonNull Database database,
                                                                                                 @NonNull String sql) {
                        requireNonNull(database);
                        requireNonNull(sql);

                        this.database = database;
                        this.originalSql = sql;
                        this.parsedSqlVariants = database.getParsedSqlVariants(sql);

                        this.bindings = new LinkedHashMap<>(Math.max(8, this.parsedSqlVariants.distinctParameterNames.size()));
                        this.preparedStatementCustomizer = null;
                        this.queryTimeout = null;
                        this.fetchSize = null;
                        this.maxRows = null;
                        this.batchChunkSize = null;
                }

                @NonNull
                @Override
                public Query bind(@NonNull String name,
                                                                                        @Nullable Object value) {
                        requireNonNull(name);

                        if (!this.parsedSqlVariants.distinctParameterNames.contains(name))
                                throw new IllegalArgumentException(format("Unknown named parameter '%s' for SQL: %s", name, this.originalSql));

                        this.bindings.put(name, value);
                        return this;
                }

                @NonNull
                @Override
                public Query bindAll(@NonNull Map<@NonNull String, @Nullable Object> parameters) {
                        requireNonNull(parameters);

                        for (Map.Entry<@NonNull String, @Nullable Object> entry : parameters.entrySet())
                                bind(entry.getKey(), entry.getValue());

                        return this;
                }

                @NonNull
                @Override
                public Query id(@Nullable Object id) {
                        this.id = id;
                        return this;
                }

                @NonNull
                @Override
                public Query queryTimeout(@Nullable Duration queryTimeout) {
                        this.queryTimeout = validateQueryTimeout(queryTimeout);
                        return this;
                }

                @NonNull
                @Override
                public Query fetchSize(@Nullable Integer fetchSize) {
                        this.fetchSize = validateNonNegativeStatementSetting("fetchSize", fetchSize);
                        return this;
                }

                @NonNull
                @Override
                public Query maxRows(@Nullable Integer maxRows) {
                        this.maxRows = validateNonNegativeStatementSetting("maxRows", maxRows);
                        return this;
                }

                @NonNull
                @Override
                public Query batchChunkSize(@Nullable Integer batchChunkSize) {
                        this.batchChunkSize = validatePositiveQuerySetting("batchChunkSize", batchChunkSize);
                        return this;
                }

                @NonNull
                @Override
                public Query resultSetMapper(@Nullable ResultSetMapper resultSetMapper) {
                        this.resultSetMapper = resultSetMapper;
                        return this;
                }

                @NonNull
                @Override
                public Query preparedStatementBinder(@Nullable PreparedStatementBinder preparedStatementBinder) {
                        this.preparedStatementBinder = preparedStatementBinder;
                        return this;
                }

                private StatementContext.@Nullable SpiOverrides spiOverrides() {
                        return this.resultSetMapper == null && this.preparedStatementBinder == null
                                        ? null
                                        : new StatementContext.SpiOverrides(this.resultSetMapper, this.preparedStatementBinder);
                }

                @NonNull
                @Override
                public Query customize(@NonNull PreparedStatementCustomizer preparedStatementCustomizer) {
                        requireNonNull(preparedStatementCustomizer);
                        this.preparedStatementCustomizer = preparedStatementCustomizer;

                        return this;
                }

                @NonNull
                @Override
                public <T> Optional<T> fetchObject(@NonNull Class<T> resultType) {
                        validateBatchChunkSizeNotSet();
                        requireNonNull(resultType);
                        PreparedQuery preparedQuery = prepare(this.bindings);
                        return this.database.queryForObject(preparedQuery.statement, resultType, effectivePreparedStatementCustomizer(), spiOverrides(), preparedQuery.parameters);
                }

                @NonNull
                @Override
                public <T> List<@Nullable T> fetchList(@NonNull Class<T> resultType) {
                        validateBatchChunkSizeNotSet();
                        requireNonNull(resultType);
                        PreparedQuery preparedQuery = prepare(this.bindings);
                        return this.database.queryForList(preparedQuery.statement, resultType, effectivePreparedStatementCustomizer(), spiOverrides(), preparedQuery.parameters);
                }

                @Nullable
                @Override
                public <T, R> R fetchStream(@NonNull Class<T> resultType,
                                                                                                                                @NonNull Function<Stream<@Nullable T>, R> streamFunction) {
                        validateBatchChunkSizeNotSet();
                        requireNonNull(resultType);
                        requireNonNull(streamFunction);
                        PreparedQuery preparedQuery = prepare(this.bindings);
                        return this.database.queryForStream(preparedQuery.statement, resultType, effectivePreparedStatementCustomizer(),
                                        this.fetchSize != null, streamFunction, spiOverrides(), preparedQuery.parameters);
                }


                @NonNull
                @Override
                public Long execute() {
                        validateBatchChunkSizeNotSet();
                        PreparedQuery preparedQuery = prepare(this.bindings);
                        return this.database.execute(preparedQuery.statement, effectivePreparedStatementCustomizer(), spiOverrides(), preparedQuery.parameters);
                }

                @NonNull
                @Override
                public <T> Optional<T> executeReturningGeneratedKey(@NonNull Class<T> resultType) {
                        validateBatchChunkSizeNotSet();
                        requireNonNull(resultType);
                        PreparedQuery preparedQuery = prepare(this.bindings);
                        return this.database.executeReturningGeneratedKey(preparedQuery.statement, resultType,
                                        effectivePreparedStatementCustomizer(), new String[0], spiOverrides(), preparedQuery.parameters);
                }

                @NonNull
                @Override
                public <T> Optional<T> executeReturningGeneratedKey(@NonNull Class<T> resultType,
                                                                                                                                                                                                                                @NonNull String @NonNull ... keyColumnNames) {
                        validateBatchChunkSizeNotSet();
                        requireNonNull(resultType);
                        PreparedQuery preparedQuery = prepare(this.bindings);
                        return this.database.executeReturningGeneratedKey(preparedQuery.statement, resultType,
                                        effectivePreparedStatementCustomizer(), keyColumnNames, spiOverrides(), preparedQuery.parameters);
                }

                @NonNull
                @Override
                public <T> List<@Nullable T> executeReturningGeneratedKeys(@NonNull Class<T> resultType) {
                        validateBatchChunkSizeNotSet();
                        requireNonNull(resultType);
                        PreparedQuery preparedQuery = prepare(this.bindings);
                        return this.database.executeReturningGeneratedKeys(preparedQuery.statement, resultType,
                                        effectivePreparedStatementCustomizer(), new String[0], spiOverrides(), preparedQuery.parameters);
                }

                @NonNull
                @Override
                public <T> List<@Nullable T> executeReturningGeneratedKeys(@NonNull Class<T> resultType,
                                                                                                                                                                                                                                         @NonNull String @NonNull ... keyColumnNames) {
                        validateBatchChunkSizeNotSet();
                        requireNonNull(resultType);
                        PreparedQuery preparedQuery = prepare(this.bindings);
                        return this.database.executeReturningGeneratedKeys(preparedQuery.statement, resultType,
                                        effectivePreparedStatementCustomizer(), keyColumnNames, spiOverrides(), preparedQuery.parameters);
                }

                @NonNull
                @Override
                public List<Long> executeBatch(@NonNull List<@NonNull Map<@NonNull String, @Nullable Object>> parameterGroups) {
                        requireNonNull(parameterGroups);
                        if (parameterGroups.isEmpty())
                                return List.of();

                        List<List<Object>> parametersAsList = new ArrayList<>(parameterGroups.size());
                        Object statementId = this.id == null ? this.database.generateId() : this.id;
                        Statement statement = null;
                        String expandedSql = null;

                        for (Map<@NonNull String, @Nullable Object> parameterGroup : parameterGroups) {
                                requireNonNull(parameterGroup);

                                for (String parameterName : parameterGroup.keySet())
                                        if (!this.parsedSqlVariants.distinctParameterNames.contains(parameterName))
                                                throw new IllegalArgumentException(format("Unknown named parameter '%s' for SQL: %s", parameterName, this.originalSql));

                                Map<String, Object> mergedBindings;
                                if (this.bindings.isEmpty()) {
                                        mergedBindings = parameterGroup;
                                } else if (parameterGroup.isEmpty()) {
                                        mergedBindings = this.bindings;
                                } else {
                                        Map<String, Object> combinedBindings = new LinkedHashMap<>(this.bindings);
                                        combinedBindings.putAll(parameterGroup);
                                        mergedBindings = combinedBindings;
                                }

                                PreparedQuery preparedQuery = prepare(mergedBindings, statementId);

                                if (expandedSql == null) {
                                        expandedSql = preparedQuery.statement.getSql();
                                        statement = preparedQuery.statement;
                                } else if (!expandedSql.equals(preparedQuery.statement.getSql())) {
                                        throw new IllegalArgumentException(format(
                                                        "Inconsistent SQL after expanding parameters for batch execution; ensure collection sizes are consistent. SQL: %s",
                                                        this.originalSql));
                                }

                                parametersAsList.add(Arrays.asList(preparedQuery.parameters));
                        }

                        if (statement == null)
                                statement = Statement.of(statementId, buildPlaceholderSql());

                        return this.database.executeBatch(statement, parametersAsList, effectivePreparedStatementCustomizer(), this.batchChunkSize, spiOverrides());
                }

                @NonNull
                @Override
                public <T> Optional<T> executeForObject(@NonNull Class<T> resultType) {
                        validateBatchChunkSizeNotSet();
                        requireNonNull(resultType);
                        PreparedQuery preparedQuery = prepare(this.bindings);
                        return this.database.executeForObject(preparedQuery.statement, resultType, effectivePreparedStatementCustomizer(), spiOverrides(), preparedQuery.parameters);
                }

                @NonNull
                @Override
                public <T> List<@Nullable T> executeForList(@NonNull Class<T> resultType) {
                        validateBatchChunkSizeNotSet();
                        requireNonNull(resultType);
                        PreparedQuery preparedQuery = prepare(this.bindings);
                        return this.database.executeForList(preparedQuery.statement, resultType, effectivePreparedStatementCustomizer(), spiOverrides(), preparedQuery.parameters);
                }

                private void validateBatchChunkSizeNotSet() {
                        if (this.batchChunkSize != null)
                                throw new IllegalStateException("batchChunkSize applies only to executeBatch(...)");
                }

                @Nullable
                private PreparedStatementCustomizer effectivePreparedStatementCustomizer() {
                        if (!this.database.hasDefaultPreparedStatementSettings()
                                        && !hasQueryPreparedStatementSettings()
                                        && this.preparedStatementCustomizer == null)
                                return null;

                        return (statementContext, preparedStatement) -> {
                                this.database.applyDefaultPreparedStatementSettings(preparedStatement);
                                applyPreparedStatementSettings(preparedStatement, this.queryTimeout, this.fetchSize, this.maxRows);

                                if (this.preparedStatementCustomizer != null)
                                        this.preparedStatementCustomizer.customize(statementContext, preparedStatement);
                        };
                }

                private boolean hasQueryPreparedStatementSettings() {
                        return this.queryTimeout != null || this.fetchSize != null || this.maxRows != null;
                }

                @NonNull
                private PreparedQuery prepare(@NonNull Map<String, Object> bindings) {
                        Object statementId = this.id == null ? this.database.generateId() : this.id;
                        return prepare(bindings, statementId);
                }

                @NonNull
                private PreparedQuery prepare(@NonNull Map<String, Object> bindings,
                                                                                                                                        @NonNull Object statementId) {
                        requireNonNull(bindings);
                        requireNonNull(statementId);

                        ParsedSql parsedSql = parsedSqlForDatabaseType();
                        List<String> parameterNames = parsedSql.parameterNames;
                        List<String> sqlFragments = sqlFragmentsForDatabaseType(parsedSql);
                        validateBindingsForParsedSql(bindings, parsedSql);

                        if (parameterNames.isEmpty())
                                return new PreparedQuery(Statement.of(statementId, sqlFragments.get(0)), new Object[0]);

                        StringBuilder sql = new StringBuilder(this.originalSql.length() + parameterNames.size() * 2);
                        List<String> missingParameterNames = null;
                        List<Object> parameters = new ArrayList<>(parameterNames.size());

                        for (int i = 0; i < parameterNames.size(); ++i) {
                                String parameterName = parameterNames.get(i);
                                sql.append(sqlFragments.get(i));

                                if (!bindings.containsKey(parameterName)) {
                                        if (missingParameterNames == null)
                                                missingParameterNames = new ArrayList<>();

                                        missingParameterNames.add(parameterName);
                                        sql.append('?');
                                        continue;
                                }

                                SecureParameterSupport.SecureParameterUnwrapResult secureParameterUnwrapResult =
                                                SecureParameterSupport.unwrapSecureAndOptionalParameterWithMetadata(bindings.get(parameterName));
                                SecureParameter secureParameter = secureParameterUnwrapResult.secureParameter();
                                Object value = secureParameterUnwrapResult.value();

                                if (value instanceof InListParameter inListParameter) {
                                        Object[] elements = inListParameter.getElements();

                                        if (elements.length == 0)
                                                throw new IllegalArgumentException(format("IN-list parameter '%s' for SQL: %s is empty", parameterName, this.originalSql));

                                        appendPlaceholders(sql, elements.length);

                                        for (int j = 0; j < elements.length; ++j) {
                                                SecureParameterSupport.SecureParameterUnwrapResult elementUnwrapResult =
                                                                SecureParameterSupport.unwrapSecureAndOptionalParameterWithMetadata(elements[j]);
                                                Object element = elementUnwrapResult.value();

                                                if (element == null)
                                                        throw new IllegalArgumentException(format(
                                                                        "IN-list parameter '%s' for SQL: %s contains null element at index %d. "
                                                                                        + "SQL IN does not match NULL values; use an explicit IS NULL predicate instead.",
                                                                        parameterName, this.originalSql, j));

                                                SecureParameter effectiveSecureParameter = secureParameter == null
                                                                ? elementUnwrapResult.secureParameter()
                                                                : secureParameter;
                                                parameters.add(effectiveSecureParameter == null
                                                                ? element
                                                                : Parameters.secure(element, SecureParameterSupport.maskOf(effectiveSecureParameter)));
                                        }
                                } else if (value instanceof Collection<?>) {
                                        throw new IllegalArgumentException(format(
                                                        "Collection parameter '%s' for SQL: %s must be wrapped with %s.inList(...) or %s.listOf/%s.setOf(...)",
                                                        parameterName, this.originalSql,
                                                        Parameters.class.getSimpleName(),
                                                        Parameters.class.getSimpleName(), Parameters.class.getSimpleName()));
                                } else if (value != null && value.getClass().isArray() && !(value instanceof byte[])) {
                                        throw new IllegalArgumentException(format(
                                                        "Array parameter '%s' for SQL: %s must be wrapped with %s.inList(...), %s.sqlArrayOf(...), or %s.arrayOf(Class, ...)",
                                                        parameterName, this.originalSql,
                                                        Parameters.class.getSimpleName(), Parameters.class.getSimpleName(), Parameters.class.getSimpleName()));
                                } else {
                                        sql.append('?');
                                        parameters.add(secureParameter == null ? value : secureParameter);
                                }
                        }

                        sql.append(sqlFragments.get(sqlFragments.size() - 1));

                        if (missingParameterNames != null)
                                throw new IllegalArgumentException(format("Missing required named parameters %s for SQL: %s", missingParameterNames, this.originalSql));

                        return new PreparedQuery(Statement.of(statementId, sql.toString()), parameters.toArray());
                }

                @NonNull
                private String buildPlaceholderSql() {
                        ParsedSql parsedSql = parsedSqlForDatabaseType();
                        List<String> parameterNames = parsedSql.parameterNames;
                        List<String> sqlFragments = sqlFragmentsForDatabaseType(parsedSql);

                        if (parameterNames.isEmpty())
                                return sqlFragments.get(0);

                        StringBuilder sql = new StringBuilder(this.originalSql.length() + parameterNames.size() * 2);

                        for (int i = 0; i < parameterNames.size(); ++i)
                                sql.append(sqlFragments.get(i)).append('?');

                        sql.append(sqlFragments.get(sqlFragments.size() - 1));
                        return sql.toString();
                }

                @NonNull
                private List<String> sqlFragmentsForDatabaseType(@NonNull ParsedSql parsedSql) {
                        requireNonNull(parsedSql);

                        if (!parsedSql.hasQuestionMarkOperators)
                                return parsedSql.sqlFragments;

                        return this.database.getDatabaseDialect().sqlFragmentsForOperators(
                                        parsedSql.hasQuestionMarkOperators,
                                        parsedSql.sqlFragments,
                                        parsedSql.questionMarkOperatorFragmentIndexes);
                }

                @NonNull
                private ParsedSql parsedSqlForDatabaseType() {
                        if (!this.parsedSqlVariants.requiresDatabaseType)
                                return this.parsedSqlVariants.standardParsedSql;

                        DatabaseType databaseType;
                        Optional<Transaction> transaction = this.database.currentTransactionForDatabaseOperation();

                        if (transaction.isPresent())
                                databaseType = this.database.getDatabaseType(transaction.get().getConnection());
                        else
                                databaseType = this.database.getDatabaseType();

                        if (databaseType == DatabaseType.MYSQL || databaseType == DatabaseType.MARIA_DB) {
                                if (this.parsedSqlVariants.mysqlParseFailureMessage != null)
                                        throw new IllegalArgumentException(this.parsedSqlVariants.mysqlParseFailureMessage);

                                return this.parsedSqlVariants.mysqlParsedSql;
                        }

                        if (this.parsedSqlVariants.standardParseFailureMessage != null)
                                throw new IllegalArgumentException(this.parsedSqlVariants.standardParseFailureMessage);

                        return this.parsedSqlVariants.standardParsedSql;
                }

                private void validateBindingsForParsedSql(@NonNull Map<String, Object> bindings,
                                                                                                                                                                         @NonNull ParsedSql parsedSql) {
                        requireNonNull(bindings);
                        requireNonNull(parsedSql);

                        for (String bindingName : bindings.keySet())
                                if (!parsedSql.distinctParameterNames.contains(bindingName))
                                        throw new IllegalArgumentException(format(
                                                        "Named parameter '%s' is not valid for the detected database SQL syntax. SQL: %s",
                                                        bindingName, this.originalSql));
                }

                private void appendPlaceholders(@NonNull StringBuilder sql,
                                                                                                                                                int count) {
                        requireNonNull(sql);

                        for (int i = 0; i < count; ++i) {
                                if (i > 0)
                                        sql.append(", ");
                                sql.append('?');
                        }
                }

                private static final class PreparedQuery {
                        @NonNull
                        private final Statement statement;
                        @NonNull
                        private final Object @NonNull [] parameters;

                        private PreparedQuery(@NonNull Statement statement,
                                                                                                                Object @NonNull [] parameters) {
                                this.statement = requireNonNull(statement);
                                this.parameters = requireNonNull(parameters);
                        }
                }

        }

        static final class ParsedSql {
                @NonNull
                private final List<String> sqlFragments;
                @NonNull
                private final List<@NonNull List<@NonNull Integer>> questionMarkOperatorFragmentIndexes;
                private final boolean hasQuestionMarkOperators;
                @NonNull
                private final List<String> parameterNames;
                @NonNull
                private final Set<String> distinctParameterNames;

                private ParsedSql(@NonNull List<String> sqlFragments,
                                                                                        @NonNull List<@NonNull List<@NonNull Integer>> questionMarkOperatorFragmentIndexes,
                                                                                        @NonNull List<String> parameterNames,
                                                                                        @NonNull Set<String> distinctParameterNames) {
                        requireNonNull(sqlFragments);
                        requireNonNull(questionMarkOperatorFragmentIndexes);
                        requireNonNull(parameterNames);
                        requireNonNull(distinctParameterNames);

                        this.sqlFragments = sqlFragments;
                        this.questionMarkOperatorFragmentIndexes = questionMarkOperatorFragmentIndexes.stream()
                                        .map(List::copyOf)
                                        .toList();
                        this.hasQuestionMarkOperators = questionMarkOperatorFragmentIndexes.stream().anyMatch(indexes -> !indexes.isEmpty());
                        this.parameterNames = parameterNames;
                        this.distinctParameterNames = distinctParameterNames;
                }

                private boolean equivalentTo(@NonNull ParsedSql other) {
                        requireNonNull(other);
                        return this.sqlFragments.equals(other.sqlFragments)
                                        && this.questionMarkOperatorFragmentIndexes.equals(other.questionMarkOperatorFragmentIndexes)
                                        && this.parameterNames.equals(other.parameterNames);
                }
        }

        enum SqlLexicalMode {
                STANDARD,
                MYSQL
        }

        @NonNull
        static ParsedSql parseNamedParameterSql(@NonNull String sql) {
                return parseNamedParameterSql(sql, SqlLexicalMode.STANDARD);
        }

        @NonNull
        static ParsedSql parseNamedParameterSql(@NonNull String sql,
                                                                                                                                                                        @NonNull SqlLexicalMode lexicalMode) {
                requireNonNull(sql);
                requireNonNull(lexicalMode);

                List<String> sqlFragments = new ArrayList<>();
                StringBuilder sqlFragment = new StringBuilder(sql.length());
                List<List<Integer>> questionMarkOperatorFragmentIndexes = new ArrayList<>();
                List<Integer> currentQuestionMarkOperatorIndexes = new ArrayList<>();
                List<String> parameterNames = new ArrayList<>();
                Set<String> distinctParameterNames = new HashSet<>();

                boolean inSingleQuote = false;
                boolean inSingleQuoteEscapesBackslash = false;
                int singleQuoteStartIndex = -1;
                boolean inDoubleQuote = false;
                int doubleQuoteStartIndex = -1;
                boolean inBacktickQuote = false;
                int backtickQuoteStartIndex = -1;
                boolean inBracketQuote = false;
                int bracketQuoteStartIndex = -1;
                boolean inLineComment = false;
                int blockCommentDepth = 0;
                int blockCommentStartIndex = -1;
                String dollarQuoteDelimiter = null;
                int dollarQuoteStartIndex = -1;
                Character oracleQuoteClosingDelimiter = null;
                int oracleQuoteStartIndex = -1;
                int previousMeaningfulIndex = -1;

                for (int i = 0; i < sql.length(); ) {
                        if (oracleQuoteClosingDelimiter != null) {
                                char c = sql.charAt(i);
                                sqlFragment.append(c);

                                if (c == oracleQuoteClosingDelimiter && i + 1 < sql.length() && sql.charAt(i + 1) == '\'') {
                                        sqlFragment.append('\'');
                                        i += 2;
                                        oracleQuoteClosingDelimiter = null;
                                        oracleQuoteStartIndex = -1;
                                        previousMeaningfulIndex = i - 1;
                                } else {
                                        ++i;
                                }

                                continue;
                        }

                        if (dollarQuoteDelimiter != null) {
                                if (sql.startsWith(dollarQuoteDelimiter, i)) {
                                        sqlFragment.append(dollarQuoteDelimiter);
                                        previousMeaningfulIndex = i + dollarQuoteDelimiter.length() - 1;
                                        i += dollarQuoteDelimiter.length();
                                        dollarQuoteDelimiter = null;
                                        dollarQuoteStartIndex = -1;
                                } else {
                                        sqlFragment.append(sql.charAt(i));
                                        ++i;
                                }

                                continue;
                        }

                        char c = sql.charAt(i);

                        if (inLineComment) {
                                sqlFragment.append(c);
                                ++i;

                                if (c == '\n' || c == '\r')
                                        inLineComment = false;

                                continue;
                        }

                        if (blockCommentDepth > 0) {
                                if (c == '/' && i + 1 < sql.length() && sql.charAt(i + 1) == '*') {
                                        sqlFragment.append("/*");
                                        i += 2;
                                        ++blockCommentDepth;
                                } else if (c == '*' && i + 1 < sql.length() && sql.charAt(i + 1) == '/') {
                                        sqlFragment.append("*/");
                                        i += 2;
                                        --blockCommentDepth;
                                        if (blockCommentDepth == 0)
                                                blockCommentStartIndex = -1;
                                } else {
                                        sqlFragment.append(c);
                                        ++i;
                                }

                                continue;
                        }

                        if (inSingleQuote) {
                                sqlFragment.append(c);

                                if (inSingleQuoteEscapesBackslash && c == '\\' && i + 1 < sql.length()) {
                                        sqlFragment.append(sql.charAt(i + 1));
                                        i += 2;
                                        continue;
                                }

                                if (c == '\'') {
                                        // Escaped quote: ''
                                        if (i + 1 < sql.length() && sql.charAt(i + 1) == '\'') {
                                                sqlFragment.append('\'');
                                                i += 2;
                                                continue;
                                        }

                                        inSingleQuote = false;
                                        inSingleQuoteEscapesBackslash = false;
                                        singleQuoteStartIndex = -1;
                                        previousMeaningfulIndex = i;
                                }

                                ++i;
                                continue;
                        }

                        if (inDoubleQuote) {
                                sqlFragment.append(c);

                                if (c == '"') {
                                        // Escaped quote: ""
                                        if (i + 1 < sql.length() && sql.charAt(i + 1) == '"') {
                                                sqlFragment.append('"');
                                                i += 2;
                                                continue;
                                        }

                                        inDoubleQuote = false;
                                        doubleQuoteStartIndex = -1;
                                        previousMeaningfulIndex = i;
                                }

                                ++i;
                                continue;
                        }

                        if (inBacktickQuote) {
                                sqlFragment.append(c);

                                if (c == '`') {
                                        inBacktickQuote = false;
                                        backtickQuoteStartIndex = -1;
                                        previousMeaningfulIndex = i;
                                }

                                ++i;
                                continue;
                        }

                        if (inBracketQuote) {
                                sqlFragment.append(c);

                                if (c == ']' && i + 1 < sql.length() && sql.charAt(i + 1) == ']') {
                                        sqlFragment.append(']');
                                        i += 2;
                                        continue;
                                }

                                if (c == ']') {
                                        inBracketQuote = false;
                                        bracketQuoteStartIndex = -1;
                                        previousMeaningfulIndex = i;
                                }

                                ++i;
                                continue;
                        }

                        // Not inside string/comment
                        if (lexicalMode == SqlLexicalMode.MYSQL && c == '#') {
                                sqlFragment.append(c);
                                ++i;
                                inLineComment = true;
                                continue;
                        }

                        if (c == '-' && i + 1 < sql.length() && sql.charAt(i + 1) == '-') {
                                sqlFragment.append("--");
                                i += 2;
                                inLineComment = true;
                                continue;
                        }

                        if (c == '/' && i + 1 < sql.length() && sql.charAt(i + 1) == '*') {
                                sqlFragment.append("/*");
                                i += 2;
                                blockCommentDepth = 1;
                                blockCommentStartIndex = i - 2;
                                continue;
                        }

                        if ((c == 'U' || c == 'u') && !isIdentifierContinuation(sql, i)
                                        && i + 2 < sql.length() && sql.charAt(i + 1) == '&' && sql.charAt(i + 2) == '\'') {
                                inSingleQuote = true;
                                inSingleQuoteEscapesBackslash = true;
                                singleQuoteStartIndex = i;
                                sqlFragment.append(c).append("&'");
                                i += 3;
                                continue;
                        }

                        if ((c == 'E' || c == 'e') && !isIdentifierContinuation(sql, i)
                                        && i + 1 < sql.length() && sql.charAt(i + 1) == '\'') {
                                inSingleQuote = true;
                                inSingleQuoteEscapesBackslash = true;
                                singleQuoteStartIndex = i;
                                sqlFragment.append(c).append('\'');
                                i += 2;
                                continue;
                        }

                        if ((c == 'Q' || c == 'q') && !isIdentifierContinuation(sql, i)
                                        && i + 2 < sql.length() && sql.charAt(i + 1) == '\''
                                        && !Character.isWhitespace(sql.charAt(i + 2)) && sql.charAt(i + 2) != '\'') {
                                char openingDelimiter = sql.charAt(i + 2);
                                oracleQuoteClosingDelimiter = oracleQuoteClosingDelimiter(openingDelimiter);
                                oracleQuoteStartIndex = i;
                                sqlFragment.append(c).append('\'').append(openingDelimiter);
                                i += 3;
                                continue;
                        }

                        if (c == '\'') {
                                inSingleQuote = true;
                                inSingleQuoteEscapesBackslash = lexicalMode == SqlLexicalMode.MYSQL;
                                singleQuoteStartIndex = i;
                                sqlFragment.append(c);
                                ++i;
                                continue;
                        }

                        if (c == '"') {
                                inDoubleQuote = true;
                                doubleQuoteStartIndex = i;
                                sqlFragment.append(c);
                                ++i;
                                continue;
                        }

                        if (c == '`') {
                                inBacktickQuote = true;
                                backtickQuoteStartIndex = i;
                                sqlFragment.append(c);
                                ++i;
                                continue;
                        }

                        if (c == '[' && !isBracketSubscriptStart(sql, i)) {
                                inBracketQuote = true;
                                bracketQuoteStartIndex = i;
                                sqlFragment.append(c);
                                ++i;
                                continue;
                        }

                        if (c == '$' && !isIdentifierContinuation(sql, i)) {
                                String delimiter = parseDollarQuoteDelimiter(sql, i);

                                if (delimiter != null) {
                                        sqlFragment.append(delimiter);
                                        i += delimiter.length();
                                        dollarQuoteDelimiter = delimiter;
                                        dollarQuoteStartIndex = i - delimiter.length();
                                        continue;
                                }
                        }

                        if (c == '?') {
                                if (isAllowedQuestionMarkOperator(sql, i, previousMeaningfulIndex)) {
                                        currentQuestionMarkOperatorIndexes.add(sqlFragment.length());
                                        sqlFragment.append(c);
                                        ++i;
                                        continue;
                                }

                                throw new IllegalArgumentException(format("Positional parameters ('?') are not supported. Use named parameters (e.g. ':id') and %s#bind. SQL: %s",
                                                Query.class.getSimpleName(), sql));
                        }

                        if (c == ':' && i + 1 < sql.length() && sql.charAt(i + 1) == ':') {
                                // Postgres type-cast operator (::), do not treat second ':' as a parameter prefix.
                                sqlFragment.append("::");
                                i += 2;
                                continue;
                        }

                        if (c == ':' && i + 1 < sql.length() && Character.isJavaIdentifierStart(sql.charAt(i + 1))) {
                                int nameStartIndex = i + 1;
                                int nameEndIndex = nameStartIndex + 1;

                                while (nameEndIndex < sql.length() && Character.isJavaIdentifierPart(sql.charAt(nameEndIndex)))
                                        ++nameEndIndex;

                                String parameterName = sql.substring(nameStartIndex, nameEndIndex);
                                parameterNames.add(parameterName);
                                distinctParameterNames.add(parameterName);
                                sqlFragments.add(sqlFragment.toString());
                                questionMarkOperatorFragmentIndexes.add(List.copyOf(currentQuestionMarkOperatorIndexes));
                                currentQuestionMarkOperatorIndexes.clear();
                                sqlFragment.setLength(0);
                                i = nameEndIndex;
                                previousMeaningfulIndex = nameEndIndex - 1;
                                continue;
                        }

                        sqlFragment.append(c);
                        if (!Character.isWhitespace(c))
                                previousMeaningfulIndex = i;
                        ++i;
                }

                validateParserTerminalState(sql, inSingleQuote, singleQuoteStartIndex, inDoubleQuote, doubleQuoteStartIndex,
                                inBacktickQuote, backtickQuoteStartIndex, inBracketQuote, bracketQuoteStartIndex,
                                blockCommentDepth, blockCommentStartIndex, dollarQuoteDelimiter, dollarQuoteStartIndex,
                                oracleQuoteClosingDelimiter, oracleQuoteStartIndex);

                sqlFragments.add(sqlFragment.toString());
                questionMarkOperatorFragmentIndexes.add(List.copyOf(currentQuestionMarkOperatorIndexes));

                return new ParsedSql(List.copyOf(sqlFragments), List.copyOf(questionMarkOperatorFragmentIndexes),
                                List.copyOf(parameterNames), Set.copyOf(distinctParameterNames));
        }

        private static char oracleQuoteClosingDelimiter(char openingDelimiter) {
                return switch (openingDelimiter) {
                        case '[' -> ']';
                        case '{' -> '}';
                        case '(' -> ')';
                        case '<' -> '>';
                        default -> openingDelimiter;
                };
        }

        @Nullable
        private static String parseDollarQuoteDelimiter(@NonNull String sql,
                                                                                                                                                                                                        int startIndex) {
                requireNonNull(sql);

                if (startIndex < 0 || startIndex >= sql.length())
                        return null;

                if (sql.charAt(startIndex) != '$')
                        return null;

                int i = startIndex + 1;

                if (i >= sql.length())
                        return null;

                char firstTagCharacter = sql.charAt(i);

                if (firstTagCharacter == '$')
                        return "$$";

                if (!isDollarQuoteTagStart(firstTagCharacter))
                        return null;

                ++i;

                while (i < sql.length()) {
                        char c = sql.charAt(i);

                        if (c == '$')
                                return sql.substring(startIndex, i + 1);

                        if (!isDollarQuoteTagPart(c))
                                return null;

                        ++i;
                }

                return null;
        }

        private static boolean isDollarQuoteTagStart(char character) {
                return Character.isLetter(character) || character == '_';
        }

        private static boolean isDollarQuoteTagPart(char character) {
                return Character.isLetterOrDigit(character) || character == '_';
        }

        private static boolean isIdentifierContinuation(@NonNull String sql,
                                                                                                                                                                                                 int startIndex) {
                requireNonNull(sql);

                if (startIndex <= 0)
                        return false;

                return Character.isJavaIdentifierPart(sql.charAt(startIndex - 1));
        }

        private static boolean isBracketSubscriptStart(@NonNull String sql,
                                                                                                                                                                                                 int startIndex) {
                requireNonNull(sql);

                if (startIndex <= 0)
                        return false;

                int previousIndex = startIndex - 1;

                while (previousIndex >= 0 && Character.isWhitespace(sql.charAt(previousIndex)))
                        --previousIndex;

                if (previousIndex < 0)
                        return false;

                char previousChar = sql.charAt(previousIndex);
                if (Character.isJavaIdentifierPart(previousChar)) {
                        int tokenStartIndex = previousIndex;

                        while (tokenStartIndex > 0 && Character.isJavaIdentifierPart(sql.charAt(tokenStartIndex - 1)))
                                --tokenStartIndex;

                        String previousToken = sql.substring(tokenStartIndex, previousIndex + 1).toUpperCase(Locale.ROOT);
                        return !BRACKET_IDENTIFIER_CONTEXT_KEYWORDS.contains(previousToken);
                }

                return previousChar == ')'
                                || previousChar == ']'
                                || previousChar == '"';
        }

        private static void validateParserTerminalState(@NonNull String sql,
                                                                                                                                                                                                        boolean inSingleQuote,
                                                                                                                                                                                                        int singleQuoteStartIndex,
                                                                                                                                                                                                        boolean inDoubleQuote,
                                                                                                                                                                                                        int doubleQuoteStartIndex,
                                                                                                                                                                                                        boolean inBacktickQuote,
                                                                                                                                                                                                        int backtickQuoteStartIndex,
                                                                                                                                                                                                        boolean inBracketQuote,
                                                                                                                                                                                                        int bracketQuoteStartIndex,
                                                                                                                                                                                                        int blockCommentDepth,
                                                                                                                                                                                                        int blockCommentStartIndex,
                                                                                                                                                                                                                                                @Nullable String dollarQuoteDelimiter,
                                                                                                                                                                                                                                                int dollarQuoteStartIndex,
                                                                                                                                                                                                                                                @Nullable Character oracleQuoteClosingDelimiter,
                                                                                                                                                                                                                                                int oracleQuoteStartIndex) {
                requireNonNull(sql);

                if (inSingleQuote)
                        throw unterminatedSqlConstructException("single-quoted string", singleQuoteStartIndex, sql);
                if (inDoubleQuote)
                        throw unterminatedSqlConstructException("double-quoted identifier", doubleQuoteStartIndex, sql);
                if (inBacktickQuote)
                        throw unterminatedSqlConstructException("backtick-quoted identifier", backtickQuoteStartIndex, sql);
                if (inBracketQuote)
                        throw unterminatedSqlConstructException("bracket-quoted identifier", bracketQuoteStartIndex, sql);
                if (blockCommentDepth > 0)
                        throw unterminatedSqlConstructException("block comment", blockCommentStartIndex, sql);
                if (dollarQuoteDelimiter != null)
                        throw unterminatedSqlConstructException(format("dollar-quoted string %s", dollarQuoteDelimiter), dollarQuoteStartIndex, sql);
                if (oracleQuoteClosingDelimiter != null)
                        throw unterminatedSqlConstructException("Oracle alternative-quoted string", oracleQuoteStartIndex, sql);
        }

        @NonNull
        private static IllegalArgumentException unterminatedSqlConstructException(@NonNull String construct,
                                                                                                                                                                                                                                                                                         int startIndex,
                                                                                                                                                                                                                                                                                         @NonNull String sql) {
                requireNonNull(construct);
                requireNonNull(sql);
                return new IllegalArgumentException(format("Unterminated %s starting at index %s. SQL: %s", construct, startIndex, sql));
        }

        @NonNull
        private static final Set<@NonNull String> QUESTION_MARK_PREFIX_KEYWORDS = Set.of(
                        "SELECT", "WHERE", "AND", "OR", "ON", "HAVING", "WHEN", "THEN", "ELSE", "IN",
                        "VALUES", "SET", "RETURNING", "USING", "LIKE", "BETWEEN", "IS", "NOT", "NULL",
                        "JOIN", "FROM"
        );

        @NonNull
        private static final Set<@NonNull String> BRACKET_IDENTIFIER_CONTEXT_KEYWORDS = Set.of(
                        "SELECT", "FROM", "JOIN", "AS", "INTO", "UPDATE", "TABLE", "BY", "ORDER", "GROUP"
        );

        @NonNull
        private static final Set<@NonNull String> QUESTION_MARK_SUFFIX_KEYWORDS = Set.of(
                        "FROM", "WHERE", "AND", "OR", "GROUP", "ORDER", "HAVING", "LIMIT", "OFFSET",
                        "UNION", "EXCEPT", "INTERSECT", "RETURNING", "JOIN", "ON"
        );

        private static boolean isAllowedQuestionMarkOperator(@NonNull String sql,
                                                                                                                                                                                                                         int questionMarkIndex,
                                                                                                                                                                                                                         int previousMeaningfulIndex) {
                requireNonNull(sql);

                int previousIndex = previousMeaningfulIndex;
                int nextIndex = nextNonWhitespaceIndex(sql, questionMarkIndex + 1);

                if (previousIndex < 0 || nextIndex < 0)
                        return false;

                char previousChar = sql.charAt(previousIndex);
                char nextChar = sql.charAt(nextIndex);

                if (isOperatorBeforeQuestionMark(previousChar))
                        return false;

                if (isTerminatorAfterQuestionMark(nextChar))
                        return false;

                if (questionMarkIndex + 1 < sql.length()) {
                        char immediateNextChar = sql.charAt(questionMarkIndex + 1);
                        if (immediateNextChar == '|' || immediateNextChar == '&') {
                                if (questionMarkIndex + 2 < sql.length() && sql.charAt(questionMarkIndex + 2) == immediateNextChar)
                                        return false;

                                String previousKeyword = keywordBefore(sql, previousIndex);
                                if (previousKeyword != null && QUESTION_MARK_PREFIX_KEYWORDS.contains(previousKeyword)
                                                && !isNamedParameterKeywordBefore(sql, previousIndex, previousKeyword))
                                        return false;

                                String nextKeyword = keywordAfter(sql, nextIndex);
                                if (nextKeyword != null && QUESTION_MARK_SUFFIX_KEYWORDS.contains(nextKeyword))
                                        return false;

                                return true;
                        }
                }

                String previousKeyword = keywordBefore(sql, previousIndex);
                if (previousKeyword != null && QUESTION_MARK_PREFIX_KEYWORDS.contains(previousKeyword))
                        return false;

                String nextKeyword = keywordAfter(sql, nextIndex);
                if (nextKeyword != null && QUESTION_MARK_SUFFIX_KEYWORDS.contains(nextKeyword))
                        return false;

                return true;
        }

        private static boolean isOperatorBeforeQuestionMark(char c) {
                return switch (c) {
                        case '=', '<', '>', '!', '+', '-', '*', '/', '%', ',', '(' -> true;
                        default -> false;
                };
        }

        private static boolean isTerminatorAfterQuestionMark(char c) {
                return switch (c) {
                        case ')', ',', ';' -> true;
                        default -> false;
                };
        }

        private static int nextNonWhitespaceIndex(@NonNull String sql,
                                                                                                                                                                                int startIndex) {
                for (int i = startIndex; i < sql.length(); i++)
                        if (!Character.isWhitespace(sql.charAt(i)))
                                return i;
                return -1;
        }

        @Nullable
        private static String keywordBefore(@NonNull String sql,
                                                                                                                                                        int index) {
                char c = sql.charAt(index);
                if (!Character.isJavaIdentifierPart(c))
                        return null;

                int endIndex = index + 1;
                int startIndex = index;
                while (startIndex >= 0 && Character.isJavaIdentifierPart(sql.charAt(startIndex)))
                        --startIndex;

                return sql.substring(startIndex + 1, endIndex).toUpperCase(Locale.ROOT);
        }

        private static boolean isNamedParameterKeywordBefore(@NonNull String sql,
                                                                                                                                                                                                                         int keywordEndIndex,
                                                                                                                                                                                                                         @NonNull String keyword) {
                requireNonNull(sql);
                requireNonNull(keyword);

                int startIndex = keywordEndIndex - keyword.length() + 1;
                return startIndex > 0 && sql.charAt(startIndex - 1) == ':';
        }

        @Nullable
        private static String keywordAfter(@NonNull String sql,
                                                                                                                                                 int index) {
                char c = sql.charAt(index);
                if (!Character.isJavaIdentifierPart(c))
                        return null;

                int endIndex = index + 1;
                while (endIndex < sql.length() && Character.isJavaIdentifierPart(sql.charAt(endIndex)))
                        ++endIndex;

                return sql.substring(index, endIndex).toUpperCase(Locale.ROOT);
        }

        /**
         * Performs a SQL query that is expected to return 0 or 1 result rows.
         *
         * @param sql              the SQL query to execute
         * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled
         * @param parameters       {@link PreparedStatement} parameters, if any
         * @param <T>              the type to be returned
         * @return a single result (or no result)
         * @throws DatabaseException if > 1 row is returned
         */
        @NonNull
        private <T> Optional<T> queryForObject(@NonNull String sql,
                                                                                                                                                                 @NonNull Class<T> resultSetRowType,
                                                                                                                                                                 Object @Nullable ... parameters) {
                requireNonNull(sql);
                requireNonNull(resultSetRowType);

                return queryForObject(Statement.of(generateId(), sql), resultSetRowType, parameters);
        }

        /**
         * Performs a SQL query that is expected to return 0 or 1 result rows.
         *
         * @param statement        the SQL statement to execute
         * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled
         * @param parameters       {@link PreparedStatement} parameters, if any
         * @param <T>              the type to be returned
         * @return a single result (or no result)
         * @throws DatabaseException if > 1 row is returned
         */
        private <T> Optional<T> queryForObject(@NonNull Statement statement,
                                                                                                                                                                 @NonNull Class<T> resultSetRowType,
                                                                                                                                                                 Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);

                return queryForObject(statement, resultSetRowType, null, null, parameters);
        }

        private <T> Optional<T> queryForObject(@NonNull Statement statement,
                                                                                                                                                                 @NonNull Class<T> resultSetRowType,
                                                                                                                                                                 @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                                                                 StatementContext.@Nullable SpiOverrides spiOverrides,
                                                                                                                                                                 Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);

                ResultHolder<Optional<T>> resultHolder = new ResultHolder<>();
                StatementContext<T> statementContext = StatementContext.<T>with(statement, this)
                                .resultSetRowType(resultSetRowType)
                                .parameters(parameters)
                                .spiOverrides(spiOverrides)
                                .build();

                List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters);

                performDatabaseOperation(statementContext, parametersAsList, preparedStatementCustomizer, (PreparedStatement preparedStatement) -> {
                        long startTime = nanoTime();

                        try (ResultSet resultSet = preparedStatement.executeQuery()) {
                                Duration executionDuration = Duration.ofNanos(nanoTime() - startTime);
                                startTime = nanoTime();

                                Optional<T> result = Optional.empty();
                                long rowsReturned = 0L;

                                if (resultSet.next()) {
                                        rowsReturned = 1L;
                                        try {
                                                T value = resultSetMapperFor(statementContext).map(statementContext, resultSet, statementContext.getResultSetRowType().get(), getInstanceProvider()).orElse(null);
                                                result = Optional.ofNullable(value);
                                        } catch (SQLException e) {
                                                throw databaseExceptionWithStatementContext(statementContext,
                                                                format("Unable to map JDBC %s row to %s", ResultSet.class.getSimpleName(), statementContext.getResultSetRowType().get()), e);
                                        }

                                        if (resultSet.next())
                                                throw databaseExceptionWithStatementContext(statementContext,
                                                                "Expected 1 row in resultset but got more than 1 instead",
                                                                new IllegalStateException("Expected 1 row in resultset but got more than 1 instead"));
                                }

                                resultHolder.value = result;
                                Duration resultSetMappingDuration = Duration.ofNanos(nanoTime() - startTime);
                                StatementResult statementResult = getMetricsCollectorDispatcher().isEnabled()
                                                ? StatementResult.ofRowsReturned(rowsReturned)
                                                : StatementResult.empty();
                                return new DatabaseOperationResult(executionDuration, resultSetMappingDuration, statementResult);
                        }
                });

                return resultHolder.value;
        }

        /**
         * Performs a SQL query that is expected to return any number of result rows.
         *
         * @param sql              the SQL query to execute
         * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled
         * @param parameters       {@link PreparedStatement} parameters, if any
         * @param <T>              the type to be returned
         * @return a list of results
         */
        @NonNull
        private <T> List<@Nullable T> queryForList(@NonNull String sql,
                                                                                                                                                                                 @NonNull Class<T> resultSetRowType,
                                                                                                                                                                                 Object @Nullable ... parameters) {
                requireNonNull(sql);
                requireNonNull(resultSetRowType);

                return queryForList(Statement.of(generateId(), sql), resultSetRowType, parameters);
        }

        /**
         * Performs a SQL query that is expected to return any number of result rows.
         *
         * @param statement        the SQL statement to execute
         * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled
         * @param parameters       {@link PreparedStatement} parameters, if any
         * @param <T>              the type to be returned
         * @return a list of results
         */
        @NonNull
        private <T> List<@Nullable T> queryForList(@NonNull Statement statement,
                                                                                                                                                                                 @NonNull Class<T> resultSetRowType,
                                                                                                                                                                                 Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);

                return queryForList(statement, resultSetRowType, null, null, parameters);
        }

        private <T> List<@Nullable T> queryForList(@NonNull Statement statement,
                                                                                                                                                                                 @NonNull Class<T> resultSetRowType,
                                                                                                                                                                                 @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                                                                                 StatementContext.@Nullable SpiOverrides spiOverrides,
                                                                                                                                                                                 Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);

                List<T> list = new ArrayList<>();
                StatementContext<T> statementContext = StatementContext.<T>with(statement, this)
                                .resultSetRowType(resultSetRowType)
                                .parameters(parameters)
                                .spiOverrides(spiOverrides)
                                .build();

                List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters);

                performDatabaseOperation(statementContext, parametersAsList, preparedStatementCustomizer, (PreparedStatement preparedStatement) -> {
                        long startTime = nanoTime();

                        try (ResultSet resultSet = preparedStatement.executeQuery()) {
                                Duration executionDuration = Duration.ofNanos(nanoTime() - startTime);
                                startTime = nanoTime();

                                while (resultSet.next()) {
                                        try {
                                                T listElement = resultSetMapperFor(statementContext).map(statementContext, resultSet, statementContext.getResultSetRowType().get(), getInstanceProvider()).orElse(null);
                                                list.add(listElement);
                                        } catch (SQLException e) {
                                                throw databaseExceptionWithStatementContext(statementContext,
                                                                format("Unable to map JDBC %s row to %s", ResultSet.class.getSimpleName(), statementContext.getResultSetRowType().get()), e);
                                        }
                                }

                                Duration resultSetMappingDuration = Duration.ofNanos(nanoTime() - startTime);
                                StatementResult statementResult = getMetricsCollectorDispatcher().isEnabled()
                                                ? StatementResult.ofRowsReturned((long) list.size())
                                                : StatementResult.empty();
                                return new DatabaseOperationResult(executionDuration, resultSetMappingDuration, statementResult);
                        }
                });

                return list;
        }

        @Nullable
        private <T, R> R queryForStream(@NonNull Statement statement,
                                                                                                                                        @NonNull Class<T> resultSetRowType,
                                                                                                                                        @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                                        boolean queryFetchSizeConfigured,
                                                                                                                                        @NonNull Function<Stream<@Nullable T>, R> streamFunction,
                                                                                                                                        StatementContext.@Nullable SpiOverrides spiOverrides,
                                                                                                                                        Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);
                requireNonNull(streamFunction);

                StatementContext<T> statementContext = StatementContext.<T>with(statement, this)
                                .resultSetRowType(resultSetRowType)
                                .parameters(parameters)
                                .spiOverrides(spiOverrides)
                                .build();

                List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters);
                StreamingResultSet<T> iterator = new StreamingResultSet<>(this, statementContext, parametersAsList,
                                preparedStatementCustomizer, queryFetchSizeConfigured);

                try {
                        try (Stream<@Nullable T> stream = StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED), false)
                                        .onClose(iterator::close)) {
                                try {
                                        return streamFunction.apply(stream);
                                } catch (Throwable throwable) {
                                        iterator.callbackFailed(throwable);
                                        if (throwable instanceof RuntimeException runtimeException)
                                                throw runtimeException;
                                        if (throwable instanceof Error error)
                                                throw error;
                                        throw new RuntimeException(throwable);
                                }
                        }
                } finally {
                        iterator.emitTerminalMetrics();
                }
        }

        /**
         * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE};
         * or a SQL statement that returns nothing, such as a DDL statement.
         *
         * @param sql        the SQL to execute
         * @param parameters {@link PreparedStatement} parameters, if any
         * @return the number of rows affected by the SQL statement
         */
        @NonNull
        private Long execute(@NonNull String sql,
                                                                                         Object @Nullable ... parameters) {
                requireNonNull(sql);
                return execute(Statement.of(generateId(), sql), parameters);
        }

        /**
         * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE};
         * or a SQL statement that returns nothing, such as a DDL statement.
         *
         * @param statement  the SQL statement to execute
         * @param parameters {@link PreparedStatement} parameters, if any
         * @return the number of rows affected by the SQL statement
         */
        @NonNull
        private Long execute(@NonNull Statement statement,
                                                                                         Object @Nullable ... parameters) {
                requireNonNull(statement);

                return execute(statement, null, null, parameters);
        }

        private Long execute(@NonNull Statement statement,
                                                                                         @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                         StatementContext.@Nullable SpiOverrides spiOverrides,
                                                                                         Object @Nullable ... parameters) {
                requireNonNull(statement);

                ResultHolder<Long> resultHolder = new ResultHolder<>();
                StatementContext<Void> statementContext = StatementContext.with(statement, this)
                                .parameters(parameters)
                                .spiOverrides(spiOverrides)
                                .build();

                List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters);

                performDatabaseOperation(statementContext, parametersAsList, preparedStatementCustomizer, (PreparedStatement preparedStatement) -> {
                        long startTime = nanoTime();
                        resultHolder.value = executeUpdate(preparedStatement);

                        Duration executionDuration = Duration.ofNanos(nanoTime() - startTime);
                        StatementResult statementResult = getMetricsCollectorDispatcher().isEnabled()
                                        ? StatementResult.ofRowsAffected(resultHolder.value)
                                        : StatementResult.empty();
                        return new DatabaseOperationResult(executionDuration, null, statementResult);
                });

                return resultHolder.value;
        }

        @NonNull
        private Long executeUpdate(@NonNull PreparedStatement preparedStatement) throws SQLException {
                requireNonNull(preparedStatement);

                DatabaseOperationSupportStatus executeLargeUpdateSupported = getExecuteLargeUpdateSupported();

                // Use the appropriate "large" value if we know it.
                // If we don't know it, detect it and store it.
                if (executeLargeUpdateSupported == DatabaseOperationSupportStatus.YES)
                        return preparedStatement.executeLargeUpdate();

                if (executeLargeUpdateSupported == DatabaseOperationSupportStatus.NO)
                        return (long) preparedStatement.executeUpdate();

                // If the driver doesn't support executeLargeUpdate, then UnsupportedOperationException is thrown.
                try {
                        Long result = preparedStatement.executeLargeUpdate();
                        setExecuteLargeUpdateSupported(DatabaseOperationSupportStatus.YES);
                        return result;
                } catch (SQLFeatureNotSupportedException | UnsupportedOperationException | AbstractMethodError e) {
                        setExecuteLargeUpdateSupported(DatabaseOperationSupportStatus.NO);
                        return (long) preparedStatement.executeUpdate();
                } catch (SQLException e) {
                        if (isUnsupportedSqlFeature(e)) {
                                setExecuteLargeUpdateSupported(DatabaseOperationSupportStatus.NO);
                                return (long) preparedStatement.executeUpdate();
                        }

                        throw e;
                }
        }

        @NonNull
        private <T> Optional<T> executeReturningGeneratedKey(@NonNull Statement statement,
                                                                                                                                                                                                                         @NonNull Class<T> resultSetRowType,
                                                                                                                                                                                                                         @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                                                                                                                         @Nullable String @Nullable [] keyColumnNames,
                                                                                                                                                                                                                         StatementContext.@Nullable SpiOverrides spiOverrides,
                                                                                                                                                                                                                         Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);

                ResultHolder<Optional<T>> resultHolder = new ResultHolder<>();
                StatementContext<T> statementContext = StatementContext.<T>with(statement, this)
                                .resultSetRowType(resultSetRowType)
                                .parameters(parameters)
                                .spiOverrides(spiOverrides)
                                .build();

                List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters);
                String[] requestedKeyColumnNames = copyGeneratedKeyColumnNames(keyColumnNames);

                performDatabaseOperation(statementContext, parametersAsList, preparedStatementCustomizer, (PreparedStatement preparedStatement) -> {
                        long startTime = nanoTime();
                        Long rowsAffected = executeUpdate(preparedStatement);

                        try (ResultSet resultSet = preparedStatement.getGeneratedKeys()) {
                                Duration executionDuration = Duration.ofNanos(nanoTime() - startTime);
                                startTime = nanoTime();

                                Optional<T> result = Optional.empty();
                                long rowsReturned = 0L;

                                if (resultSet.next()) {
                                        rowsReturned = 1L;
                                        try {
                                                T value = resultSetMapperFor(statementContext).map(statementContext, resultSet, statementContext.getResultSetRowType().get(), getInstanceProvider()).orElse(null);
                                                result = Optional.ofNullable(value);
                                        } catch (SQLException e) {
                                                throw databaseExceptionWithStatementContext(statementContext,
                                                                format("Unable to map JDBC generated-key row to %s", statementContext.getResultSetRowType().get()), e);
                                        }

                                        if (resultSet.next())
                                                throw databaseExceptionWithStatementContext(statementContext,
                                                                "Expected 1 generated-key row but got more than 1 instead",
                                                                new IllegalStateException("Expected 1 generated-key row but got more than 1 instead"));
                                }

                                resultHolder.value = result;
                                Duration resultSetMappingDuration = Duration.ofNanos(nanoTime() - startTime);
                                StatementResult statementResult = getMetricsCollectorDispatcher().isEnabled()
                                                ? new StatementResult(rowsReturned, rowsAffected)
                                                : StatementResult.empty();
                                return new DatabaseOperationResult(executionDuration, resultSetMappingDuration, statementResult);
                        }
                }, generatedKeysPreparedStatementFactory(requestedKeyColumnNames));

                return resultHolder.value;
        }

        @NonNull
        private <T> List<@Nullable T> executeReturningGeneratedKeys(@NonNull Statement statement,
                                                                                                                                                                                                                                                 @NonNull Class<T> resultSetRowType,
                                                                                                                                                                                                                                                 @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                                                                                                                                                 @Nullable String @Nullable [] keyColumnNames,
                                                                                                                                                                                                                                                 StatementContext.@Nullable SpiOverrides spiOverrides,
                                                                                                                                                                                                                                                 Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);

                List<T> list = new ArrayList<>();
                StatementContext<T> statementContext = StatementContext.<T>with(statement, this)
                                .resultSetRowType(resultSetRowType)
                                .parameters(parameters)
                                .spiOverrides(spiOverrides)
                                .build();

                List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters);
                String[] requestedKeyColumnNames = copyGeneratedKeyColumnNames(keyColumnNames);

                performDatabaseOperation(statementContext, parametersAsList, preparedStatementCustomizer, (PreparedStatement preparedStatement) -> {
                        long startTime = nanoTime();
                        Long rowsAffected = executeUpdate(preparedStatement);

                        try (ResultSet resultSet = preparedStatement.getGeneratedKeys()) {
                                Duration executionDuration = Duration.ofNanos(nanoTime() - startTime);
                                startTime = nanoTime();

                                while (resultSet.next()) {
                                        try {
                                                T listElement = resultSetMapperFor(statementContext).map(statementContext, resultSet, statementContext.getResultSetRowType().get(), getInstanceProvider()).orElse(null);
                                                list.add(listElement);
                                        } catch (SQLException e) {
                                                throw databaseExceptionWithStatementContext(statementContext,
                                                                format("Unable to map JDBC generated-key row to %s", statementContext.getResultSetRowType().get()), e);
                                        }
                                }

                                Duration resultSetMappingDuration = Duration.ofNanos(nanoTime() - startTime);
                                StatementResult statementResult = getMetricsCollectorDispatcher().isEnabled()
                                                ? new StatementResult((long) list.size(), rowsAffected)
                                                : StatementResult.empty();
                                return new DatabaseOperationResult(executionDuration, resultSetMappingDuration, statementResult);
                        }
                }, generatedKeysPreparedStatementFactory(requestedKeyColumnNames));

                return list;
        }

        @NonNull
        private String[] copyGeneratedKeyColumnNames(@Nullable String @Nullable [] keyColumnNames) {
                if (keyColumnNames == null || keyColumnNames.length == 0)
                        return new String[0];

                String[] copy = Arrays.copyOf(keyColumnNames, keyColumnNames.length);

                for (String keyColumnName : copy)
                        requireNonNull(keyColumnName);

                return copy;
        }

        @NonNull
        private PreparedStatementFactory generatedKeysPreparedStatementFactory(@NonNull String @NonNull [] keyColumnNames) {
                requireNonNull(keyColumnNames);

                return (connection, statementContext) -> getDatabaseDialect(connection)
                                .prepareGeneratedKeysStatement(connection, statementContext, keyColumnNames);
        }

        /**
         * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE},
         * which returns 0 or 1 rows with database-native syntax such as PostgreSQL/SQLite {@code RETURNING},
         * MariaDB {@code INSERT ... RETURNING},
         * or SQL Server {@code OUTPUT}.
         *
         * @param sql              the SQL query to execute
         * @param resultSetRowType the type to which the {@link ResultSet} row should be marshaled
         * @param parameters       {@link PreparedStatement} parameters, if any
         * @param <T>              the type to be returned
         * @return a single result (or no result)
         * @throws DatabaseException if > 1 row is returned
         */
        @NonNull
        private <T> Optional<T> executeForObject(@NonNull String sql,
                                                                                                                                                                         @NonNull Class<T> resultSetRowType,
                                                                                                                                                                         Object @Nullable ... parameters) {
                requireNonNull(sql);
                requireNonNull(resultSetRowType);

                return executeForObject(Statement.of(generateId(), sql), resultSetRowType, parameters);
        }

        /**
         * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE},
         * which returns 0 or 1 rows with database-native syntax such as PostgreSQL/SQLite {@code RETURNING},
         * MariaDB {@code INSERT ... RETURNING},
         * or SQL Server {@code OUTPUT}.
         *
         * @param statement        the SQL statement to execute
         * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled
         * @param parameters       {@link PreparedStatement} parameters, if any
         * @param <T>              the type to be returned
         * @return a single result (or no result)
         * @throws DatabaseException if > 1 row is returned
         */
        private <T> Optional<T> executeForObject(@NonNull Statement statement,
                                                                                                                                                                         @NonNull Class<T> resultSetRowType,
                                                                                                                                                                         Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);

                return executeForObject(statement, resultSetRowType, null, null, parameters);
        }

        private <T> Optional<T> executeForObject(@NonNull Statement statement,
                                                                                                                                                                         @NonNull Class<T> resultSetRowType,
                                                                                                                                                                         @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                                                                         StatementContext.@Nullable SpiOverrides spiOverrides,
                                                                                                                                                                         Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);

                // Ultimately we just delegate to queryForObject.
                // Having `executeForList` is to allow for users to explicitly express intent
                // and make static analysis of code easier (e.g. maybe you'd like to hook all of your "execute" statements for
                // logging, or delegation to a writable master as opposed to a read replica)
                return queryForObject(statement, resultSetRowType, preparedStatementCustomizer, spiOverrides, parameters);
        }

        /**
         * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE},
         * which returns any number of rows with database-native syntax such as PostgreSQL/SQLite {@code RETURNING},
         * MariaDB {@code INSERT ... RETURNING},
         * or SQL Server {@code OUTPUT}.
         *
         * @param sql              the SQL to execute
         * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled
         * @param parameters       {@link PreparedStatement} parameters, if any
         * @param <T>              the type to be returned
         * @return a list of results
         */
        @NonNull
        private <T> List<@Nullable T> executeForList(@NonNull String sql,
                                                                                                                                                                                         @NonNull Class<T> resultSetRowType,
                                                                                                                                                                                         Object @Nullable ... parameters) {
                requireNonNull(sql);
                requireNonNull(resultSetRowType);

                return executeForList(Statement.of(generateId(), sql), resultSetRowType, parameters);
        }

        /**
         * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE},
         * which returns any number of rows with database-native syntax such as PostgreSQL/SQLite {@code RETURNING},
         * MariaDB {@code INSERT ... RETURNING},
         * or SQL Server {@code OUTPUT}.
         *
         * @param statement        the SQL statement to execute
         * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled
         * @param parameters       {@link PreparedStatement} parameters, if any
         * @param <T>              the type to be returned
         * @return a list of results
         */
        @NonNull
        private <T> List<@Nullable T> executeForList(@NonNull Statement statement,
                                                                                                                                                                                         @NonNull Class<T> resultSetRowType,
                                                                                                                                                                                         Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);

                return executeForList(statement, resultSetRowType, null, null, parameters);
        }

        private <T> List<@Nullable T> executeForList(@NonNull Statement statement,
                                                                                                                                                                                         @NonNull Class<T> resultSetRowType,
                                                                                                                                                                                         @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                                                                                         StatementContext.@Nullable SpiOverrides spiOverrides,
                                                                                                                                                                                         Object @Nullable ... parameters) {
                requireNonNull(statement);
                requireNonNull(resultSetRowType);

                // Ultimately we just delegate to queryForList.
                // Having `executeForList` is to allow for users to explicitly express intent
                // and make static analysis of code easier (e.g. maybe you'd like to hook all of your "execute" statements for
                // logging, or delegation to a writable master as opposed to a read replica)
                return queryForList(statement, resultSetRowType, preparedStatementCustomizer, spiOverrides, parameters);
        }

        /**
         * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE}
         * in "batch" over a set of parameter groups.
         * <p>
         * Useful for bulk-inserting or updating large amounts of data.
         *
         * @param sql             the SQL to execute
         * @param parameterGroups Groups of {@link PreparedStatement} parameters
         * @return the number of rows affected by the SQL statement per-group
         */
        @NonNull
        private List<Long> executeBatch(@NonNull String sql,
                                                                                                                                        @NonNull List<List<Object>> parameterGroups) {
                requireNonNull(sql);
                requireNonNull(parameterGroups);

                return executeBatch(Statement.of(generateId(), sql), parameterGroups);
        }

        /**
         * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE}
         * in "batch" over a set of parameter groups.
         * <p>
         * Useful for bulk-inserting or updating large amounts of data.
         *
         * @param statement       the SQL statement to execute
         * @param parameterGroups Groups of {@link PreparedStatement} parameters
         * @return the number of rows affected by the SQL statement per-group
         */
        @NonNull
        private List<Long> executeBatch(@NonNull Statement statement,
                                                                                                                                        @NonNull List<List<Object>> parameterGroups) {
                requireNonNull(statement);
                requireNonNull(parameterGroups);

                return executeBatch(statement, parameterGroups, null, null, null);
        }

        private List<Long> executeBatch(@NonNull Statement statement,
                                                                                                                                        @NonNull List<List<Object>> parameterGroups,
                                                                                                                                        @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                                        @Nullable Integer batchChunkSize,
                                                                                                                                        StatementContext.@Nullable SpiOverrides spiOverrides) {
                requireNonNull(statement);
                requireNonNull(parameterGroups);
                if (parameterGroups.isEmpty())
                        return List.of();

                Integer expectedParameterCount = null;

                for (int i = 0; i < parameterGroups.size(); i++) {
                        List<Object> parameterGroup = parameterGroups.get(i);

                        if (parameterGroup == null)
                                throw new IllegalArgumentException(format("Parameter group at index %s is null", i));

                        int parameterCount = parameterGroup.size();
                        if (expectedParameterCount == null) {
                                expectedParameterCount = parameterCount;
                        } else if (parameterCount != expectedParameterCount) {
                                throw new IllegalArgumentException(format(
                                                "Inconsistent parameter group size at index %s: expected %s but found %s",
                                                i, expectedParameterCount, parameterCount));
                        }
                }

                ResultHolder<List<Long>> resultHolder = new ResultHolder<>();
                StatementContext<List<Long>> statementContext = StatementContext.with(statement, this)
                                .parameters((List) parameterGroups)
                                .resultSetRowType(List.class)
                                .batchParameterGroups(true)
                                .spiOverrides(spiOverrides)
                                .build();

                if (batchChunkSize == null || batchChunkSize >= parameterGroups.size()) {
                        performDatabaseOperation(statementContext, (preparedStatement) -> {
                                applyPreparedStatementCustomizer(statementContext, preparedStatement, preparedStatementCustomizer);

                                for (List<Object> parameterGroup : parameterGroups) {
                                        if (parameterGroup.size() > 0)
                                                performPreparedStatementBinding(statementContext, preparedStatement, parameterGroup);

                                        preparedStatement.addBatch();
                                }
                        }, (PreparedStatement preparedStatement) -> {
                                long startTime = nanoTime();
                                List<Long> result = executePreparedStatementBatch(preparedStatement);

                                resultHolder.value = result;
                                return batchDatabaseOperationResult(startTime, result);
                        }, parameterGroups.size());
                } else {
                        int effectiveBatchChunkSize = batchChunkSize;

                        performDatabaseOperation(statementContext, (preparedStatement) -> {
                                applyPreparedStatementCustomizer(statementContext, preparedStatement, preparedStatementCustomizer);
                        }, (PreparedStatement preparedStatement) -> {
                                long startTime = nanoTime();
                                int currentBatchSize = 0;
                                List<Long> result = new ArrayList<>(parameterGroups.size());

                                for (List<Object> parameterGroup : parameterGroups) {
                                        if (parameterGroup.size() > 0)
                                                performPreparedStatementBinding(statementContext, preparedStatement, parameterGroup);

                                        preparedStatement.addBatch();
                                        ++currentBatchSize;

                                        if (currentBatchSize == effectiveBatchChunkSize) {
                                                result.addAll(executePreparedStatementBatch(preparedStatement));
                                                preparedStatement.clearBatch();
                                                currentBatchSize = 0;
                                        }
                                }

                                if (currentBatchSize > 0) {
                                        result.addAll(executePreparedStatementBatch(preparedStatement));
                                        preparedStatement.clearBatch();
                                }

                                resultHolder.value = result;
                                return batchDatabaseOperationResult(startTime, result);
                        }, parameterGroups.size());
                }

                return resultHolder.value;
        }

        @NonNull
        private DatabaseOperationResult batchDatabaseOperationResult(long startTime,
                                                                                                                                                                                                                                                        @NonNull List<Long> result) {
                requireNonNull(result);

                Duration executionDuration = Duration.ofNanos(nanoTime() - startTime);
                StatementResult statementResult = StatementResult.empty();
                if (getMetricsCollectorDispatcher().isEnabled()) {
                        Long rowsAffected = sumBatchUpdateCounts(result);
                        statementResult = rowsAffected == null ? StatementResult.empty() : StatementResult.ofRowsAffected(rowsAffected);
                }

                return new DatabaseOperationResult(executionDuration, null, statementResult);
        }

        @NonNull
        private List<Long> executePreparedStatementBatch(@NonNull PreparedStatement preparedStatement) throws SQLException {
                requireNonNull(preparedStatement);

                DatabaseOperationSupportStatus executeLargeBatchSupported = getExecuteLargeBatchSupported();

                // Use the appropriate "large" value if we know it.
                // If we don't know it, detect it and store it.
                if (executeLargeBatchSupported == DatabaseOperationSupportStatus.YES) {
                        long[] resultArray = preparedStatement.executeLargeBatch();
                        return Arrays.stream(resultArray).boxed().collect(Collectors.toList());
                }
                if (executeLargeBatchSupported == DatabaseOperationSupportStatus.NO) {
                        int[] resultArray = preparedStatement.executeBatch();
                        return Arrays.stream(resultArray).asLongStream().boxed().collect(Collectors.toList());
                }

                // If the driver doesn't support executeLargeBatch, then UnsupportedOperationException is thrown.
                try {
                        long[] resultArray = preparedStatement.executeLargeBatch();
                        setExecuteLargeBatchSupported(DatabaseOperationSupportStatus.YES);
                        return Arrays.stream(resultArray).boxed().collect(Collectors.toList());
                } catch (SQLFeatureNotSupportedException | UnsupportedOperationException | AbstractMethodError e) {
                        setExecuteLargeBatchSupported(DatabaseOperationSupportStatus.NO);
                        int[] resultArray = preparedStatement.executeBatch();
                        return Arrays.stream(resultArray).asLongStream().boxed().collect(Collectors.toList());
                } catch (SQLException e) {
                        if (!isUnsupportedSqlFeature(e))
                                throw e;

                        setExecuteLargeBatchSupported(DatabaseOperationSupportStatus.NO);
                        int[] resultArray = preparedStatement.executeBatch();
                        return Arrays.stream(resultArray).asLongStream().boxed().collect(Collectors.toList());
                }
        }

        /**
         * Exposes a temporary handle to JDBC {@link DatabaseMetaData}, which provides comprehensive vendor-specific information about this database as a whole.
         * <p>
         * This method acquires {@link DatabaseMetaData} on its own newly-borrowed connection, which it manages internally.
         * <p>
         * It does <strong>not</strong> participate in the active transaction, if one exists.
         * <p>
         * The connection is closed as soon as {@link DatabaseMetaDataReader#read(DatabaseMetaData)} completes.
         * <p>
         * See <a href="https://docs.oracle.com/en/java/javase/26/docs/api/java.sql/java/sql/DatabaseMetaData.html">{@code DatabaseMetaData} Javadoc</a> for details.
         */
        public void readDatabaseMetaData(@NonNull DatabaseMetaDataReader databaseMetaDataReader) {
                requireNonNull(databaseMetaDataReader);

                performRawConnectionOperation((connection -> {
                        databaseMetaDataReader.read(connection.getMetaData());
                        return Optional.empty();
                }), false);
        }

        /**
         * Performs raw JDBC work with a Pyranid-managed {@link Connection}.
         * <p>
         * If called inside a Pyranid transaction, this operation uses the transaction's connection and participates in that
         * transaction. Otherwise, Pyranid borrows a connection for the duration of the callback and closes it afterwards.
         * <p>
         * The {@link Connection} passed to {@code rawConnectionOperation} is a guarded handle. Normal JDBC operations are
         * delegated to the underlying driver connection, but lifecycle, transaction-management, and connection-wide state
         * methods such as
         * {@link Connection#close()}, {@link Connection#commit()}, {@link Connection#rollback()},
         * {@link Connection#setAutoCommit(boolean)}, {@link Connection#setCatalog(String)}, {@link Connection#setSchema(String)}, and
         * {@link Connection#setNetworkTimeout(java.util.concurrent.Executor, int)} throw {@link IllegalStateException}. Use
         * Pyranid transaction APIs instead.
         * {@link Connection#unwrap(Class)} may return a guarded, callback-scoped proxy for a vendor interface, but never a
         * castable physical {@link Connection}; the proxy blocks lifecycle methods and expires with the callback.
         * JDBC objects created from this handle are also guarded: {@link java.sql.Statement#getConnection()} and
         * {@link java.sql.DatabaseMetaData#getConnection()} return the Pyranid-managed handle, and
         * {@link ResultSet#getStatement()} returns a guarded statement. Guarded statements, resultsets, and metadata refuse
         * driver-specific {@code unwrap(...)} calls that could expose the driver's underlying connection.
         * <p>
         * The connection handle is valid only for the duration of the callback. Do not close it, retain it, or use it after this
         * method returns.
         *
         * @param rawConnectionOperation the raw JDBC operation to perform
         * @param <T>                    the type to be returned
         * @return the operation result
         * @throws DatabaseException if connection acquisition, callback execution, or cleanup fails
         * @since 4.2.0
         */
        @NonNull
        public <T> Optional<T> useRawConnection(@NonNull RawConnectionOperation<T> rawConnectionOperation) {
                requireNonNull(rawConnectionOperation);

                return performRawConnectionOperation(connection -> {
                        PyranidRawConnection rawConnection = new PyranidRawConnection(connection);

                        try {
                                Optional<T> result = rawConnectionOperation.perform(rawConnection);
                                return result == null ? Optional.empty() : result;
                        } finally {
                                rawConnection.release();
                        }
                }, true);
        }

        /**
         * Performs an operation with one callback-scoped database-notification listener session.
         * <p>
         * This method is synchronous and blocking. It acquires at most one listener connection from this
         * {@code Database}'s configured {@link DataSource}, registers every requested channel, invokes {@code operation}
         * at most once, expires the supplied {@link NotificationSession}, and completes cleanup before returning or
         * throwing. It never reconnects.
         * <p>
         * The configured source must preserve one physical backend session for the entire checkout. For PostgreSQL,
         * direct connections and session pooling are suitable. Using PgBouncer transaction or statement pooling as the
         * listener source is unsupported: registration can appear to succeed before backend-session affinity is lost and
         * notification delivery silently stops. Pyranid does not inspect or validate proxy topology. Applications whose
         * ordinary source cannot provide the required affinity should construct a separate {@code Database} over a suitable
         * listener source and invoke this method on that instance.
         * <p>
         * Notifications are lossy hints. Durable applications should normally reconcile authoritative state as the first
         * callback action. The operation may use ordinary database methods, but those methods acquire or select their
         * connection normally and never reuse the listener connection.
         * <p>
         * A terminal receive failure is retained by the session and rethrown after cleanup even if {@code operation} catches
         * it and returns. A retained transport {@link Error} propagates as that exact, unwrapped instance. If the callback
         * instead throws a distinct {@code Error}, that callback error remains primary and the retained transport failure is
         * suppressed beneath it.
         *
         * @param channels  fixed, nonempty set of nonblank channels to register
         * @param operation operation to invoke after every channel has been registered
         * @throws NullPointerException if {@code channels}, a channel, or {@code operation} is null
         * @throws IllegalArgumentException if the set is empty or a channel violates common or backend-specific limits
         * @throws IllegalStateException if any Pyranid transaction is active on the calling thread
         * @throws InterruptedException if cooperative interruption wins after any required cleanup
         * @throws UnsupportedOperationException if the database dialect or runtime driver cannot receive notifications
         * @throws DatabaseException if connection acquisition, setup, receive, callback execution, or cleanup fails
         * @since 4.6.0
         */
        public void withNotificationSession(@NonNull Set<@NonNull String> channels,
                                                                                                                                                        @NonNull NotificationSessionOperation operation)
                        throws InterruptedException {
                requireNonNull(channels);
                requireNonNull(operation);

                if (channels.isEmpty())
                        throw new IllegalArgumentException("channels must not be empty");

                Set<String> validatedChannels = new LinkedHashSet<>(channels.size());

                for (String channel : channels) {
                        Notification.validateChannel(channel);
                        validatedChannels.add(channel);
                }

                if (hasAmbientTransaction())
                        throw new IllegalStateException("Notification sessions are not permitted inside a Pyranid transaction");

                if (Thread.interrupted())
                        throw new InterruptedException();

                DatabaseType databaseType = getDatabaseType();
                DatabaseNotificationSupport notificationSupport = getDatabaseDialect().notificationSupport();

                for (String channel : validatedChannels)
                        notificationSupport.validateChannel(channel);

                if (Thread.interrupted())
                        throw new InterruptedException();

                performNotificationSession(Set.copyOf(validatedChannels), operation, databaseType, notificationSupport);
        }

        /**
         * Performs an operation with one callback-scoped database-notification listener session for a single channel.
         * <p>
         * This is the single-channel convenience form of
         * {@link #withNotificationSession(Set, NotificationSessionOperation)}. The listener connection comes from this
         * {@code Database}'s configured {@link DataSource}, which must preserve physical backend-session affinity for the
         * entire checkout. For PostgreSQL, using PgBouncer transaction or statement pooling as the listener source is
         * unsupported: registration can appear to succeed before affinity is lost and delivery silently stops. Pyranid does
         * not inspect or validate proxy topology.
         * <p>
         * A terminal receive failure is retained and rethrown after cleanup even if {@code operation} catches it and returns.
         * A retained transport {@link Error} propagates as the exact, unwrapped instance unless a distinct callback
         * {@code Error} takes precedence as described by the set-based overload.
         *
         * @param channel   nonblank channel to register
         * @param operation operation to invoke after the channel has been registered
         * @throws NullPointerException if {@code channel} or {@code operation} is null
         * @throws IllegalArgumentException if the channel violates common or backend-specific limits
         * @throws IllegalStateException if any Pyranid transaction is active on the calling thread
         * @throws InterruptedException if cooperative interruption wins after any required cleanup
         * @throws UnsupportedOperationException if the database dialect or runtime driver cannot receive notifications
         * @throws DatabaseException if connection acquisition, setup, receive, callback execution, or cleanup fails
         * @since 4.6.0
         */
        public void withNotificationSession(@NonNull String channel,
                                                                                                                                                        @NonNull NotificationSessionOperation operation)
                        throws InterruptedException {
                requireNonNull(channel);
                requireNonNull(operation);
                withNotificationSession(Set.of(channel), operation);
        }

        /**
         * Sends a transient database notification without specifying a payload.
         * <p>
         * Sending follows ordinary Pyranid statement and transaction selection. On PostgreSQL, a send inside a Pyranid
         * transaction becomes visible only if that transaction commits. Payload representation is database-specific;
         * PostgreSQL converts the resulting null payload to the empty string.
         *
         * @param channel nonblank notification channel
         * @throws NullPointerException if {@code channel} is null
         * @throws IllegalArgumentException if the channel violates common or backend-specific limits
         * @throws UnsupportedOperationException if the database dialect does not support notification sends
         * @throws DatabaseException if the send fails
         * @since 4.6.0
         */
        public void sendNotification(@NonNull String channel) {
                sendNotification(channel, null);
        }

        /**
         * Sends a transient database notification.
         * <p>
         * Sending follows ordinary Pyranid statement and transaction selection, including connection ownership,
         * parameter binding and redaction, statement logging, timeout configuration, and metrics. Payload nullability
         * and null/empty-string handling are database-specific; Pyranid performs no generic normalization.
         * <p>
         * On PostgreSQL this executes bound {@code pg_notify(?, ?)} SQL. PostgreSQL converts a null payload to the empty
         * string. Delivery occurs only after commit and is discarded by rollback; notification delivery itself remains
         * non-durable and may be coalesced.
         *
         * @param channel nonblank notification channel
         * @param payload notification payload, possibly null or empty
         * @throws NullPointerException if {@code channel} is null
         * @throws IllegalArgumentException if the channel or payload violates common or backend-specific limits
         * @throws UnsupportedOperationException if the database dialect does not support notification sends
         * @throws DatabaseException if the send fails
         * @since 4.6.0
         */
        public void sendNotification(@NonNull String channel,
                                                                                                 @Nullable String payload) {
                Notification.validateChannel(channel);
                DatabaseNotificationSupport notificationSupport = getDatabaseDialect().notificationSupport();

                notificationSupport.validateChannel(channel);
                notificationSupport.validatePayload(payload);

                if (!notificationSupport.isSendSupported())
                        throw new UnsupportedOperationException(format(
                                        "Database type %s does not support notification sends", getDatabaseType()));

                Statement statement = Statement.of(generateId(), notificationSupport.sendStatementSql());
                StatementContext<Void> statementContext = StatementContext.with(statement, this)
                                .parameters(channel, payload)
                                .build();

                PreparedStatementCustomizer defaultPreparedStatementCustomizer = hasDefaultPreparedStatementSettings()
                                ? (context, preparedStatement) -> applyDefaultPreparedStatementSettings(preparedStatement)
                                : null;

                performDatabaseOperation(statementContext, Arrays.asList(channel, payload),
                                defaultPreparedStatementCustomizer, preparedStatement -> {
                        long startTime = nanoTime();
                        executeAndDrain(preparedStatement);
                        return new DatabaseOperationResult(
                                        Duration.ofNanos(nanoTime() - startTime), null, StatementResult.empty());
                });
        }

        /**
         * Reports whether the configured database dialect and currently loadable runtime adapter expose the APIs required
         * to attempt a notification-listening session.
         * <p>
         * This method resolves the full database type. If it has not been configured or cached, resolution may acquire a
         * metadata connection and may throw {@link DatabaseException}. It does not acquire a listener session, inspect pool
         * or proxy mode, prove backend-session affinity, unwrap a physical listener connection, or emit notification-session
         * lifecycle metrics.
         *
         * @return true if notification listening can be attempted with the current dialect and runtime
         * @throws DatabaseException if automatic database-type detection fails
         * @since 4.6.0
         */
        @NonNull
        public Boolean isNotificationListeningSupported() {
                return getDatabaseDialect().notificationSupport().isReceiveRuntimeAvailable();
        }

        private void executeAndDrain(@NonNull PreparedStatement preparedStatement) throws SQLException {
                requireNonNull(preparedStatement);

                boolean resultAvailable = preparedStatement.execute();

                for (;;) {
                        if (resultAvailable) {
                                try (ResultSet resultSet = preparedStatement.getResultSet()) {
                                        if (resultSet != null) {
                                                while (resultSet.next()) {
                                                        // Drain every row before advancing to a possible subsequent result.
                                                }
                                        }
                                }
                        } else if (preparedStatement.getUpdateCount() == -1) {
                                break;
                        }

                        resultAvailable = preparedStatement.getMoreResults(java.sql.Statement.CLOSE_CURRENT_RESULT);
                }
        }

        private void performNotificationSession(@NonNull Set<@NonNull String> channels,
                                                                                                                                                                        @NonNull NotificationSessionOperation operation,
                                                                                                                                                                        @NonNull DatabaseType databaseType,
                                                                                                                                                                        @NonNull DatabaseNotificationSupport notificationSupport)
                        throws InterruptedException {
                requireNonNull(channels);
                requireNonNull(operation);
                requireNonNull(databaseType);
                requireNonNull(notificationSupport);

                MetricsCollectorDispatcher metricsCollectorDispatcher = getMetricsCollectorDispatcher();
                UUID notificationSessionId = metricsCollectorDispatcher.isEnabled() ? UUID.randomUUID() : null;
                long lifecycleStartTime = notificationSessionId == null ? 0L : nanoTime();

                if (notificationSessionId != null)
                        metricsCollectorDispatcher.willOpenNotificationSession(databaseType, notificationSessionId);

                Connection connection = null;
                NotificationTransport transport = null;
                NotificationSession session = null;
                Boolean initialAutoCommit = null;
                NotificationSetupPhase setupPhase = NotificationSetupPhase.CAPABILITY;
                UnsupportedOperationException frameworkUnsupportedFailure = null;
                boolean initialSanitationComplete = false;

                try {
                        if (!notificationSupport.isReceiveRuntimeAvailable()) {
                                frameworkUnsupportedFailure = new UnsupportedOperationException(format(
                                                "Database type %s does not support notification listening with the current runtime", databaseType));
                                throw frameworkUnsupportedFailure;
                        }

                        setupPhase = NotificationSetupPhase.ACQUIRE_CONNECTION;
                        connection = requireNonNull(getDataSource().getConnection(), "DataSource returned a null connection");

                        setupPhase = NotificationSetupPhase.CONFIGURE_CONNECTION;
                        initialAutoCommit = connection.getAutoCommit();

                        if (!initialAutoCommit) {
                                // A pooled connection can be returned with an unfinished transaction. Discard inherited work before
                                // enabling autocommit; LISTEN registration must not inherit that transaction.
                                connection.rollback();
                                connection.setAutoCommit(true);
                        }

                        setupPhase = NotificationSetupPhase.OPEN_TRANSPORT;
                        transport = notificationSupport.open(connection);

                        setupPhase = NotificationSetupPhase.SANITIZE_UNLISTEN;
                        transport.unlistenAll();

                        setupPhase = NotificationSetupPhase.SANITIZE_DRAIN;
                        requireNonNull(transport.drain(), "Notification transport returned a null batch");

                        if (transport.isConnectionUncertain())
                                throw new SQLException("Notification transport became uncertain during initial sanitation");

                        initialSanitationComplete = true;
                        setupPhase = NotificationSetupPhase.REGISTER;
                        transport.listen(channels);

                        if (transport.isConnectionUncertain())
                                throw new SQLException("Notification transport became uncertain during registration");

                        session = new NotificationSession(
                                        this, requireNonNull(transport), databaseType, notificationSessionId);
                } catch (Throwable setupCause) {
                        boolean connectionFailure = isNotificationConnectionFailure(connection, setupCause);
                        Throwable setupFailure = normalizeNotificationSetupFailure(
                                        setupCause, frameworkUnsupportedFailure, setupPhase, connectionFailure, databaseType);

                        if (notificationSessionId != null) {
                                metricsCollectorDispatcher.didFailToOpenNotificationSession(
                                                databaseType, notificationSessionId,
                                                Duration.ofNanos(nanoTime() - lifecycleStartTime), setupFailure);
                        }

                        boolean interruptionObserved = Thread.interrupted();
                        // Synthetic uncertainty failures need not carry a connection-class SQLSTATE. Cleanup still fails closed:
                        // this independent transport-state check prevents reuse even if generic exception classification does not.
                        boolean canAttemptHealthyCleanup = initialSanitationComplete
                                        && connection != null
                                        && transport != null
                                        && initialAutoCommit != null
                                        && setupPhase == NotificationSetupPhase.REGISTER
                                        && !connectionFailure
                                        && !transport.isConnectionUncertain();
                        Throwable cleanupFailure = canAttemptHealthyCleanup
                                        ? cleanupOpenedNotificationSession(connection, transport, initialAutoCommit, false, databaseType)
                                        : cleanupFailedNotificationCandidate(connection, databaseType);
                        setupFailure = appendSuppressedIfPresent(setupFailure, cleanupFailure);

                        if (Thread.interrupted())
                                interruptionObserved = true;

                        if (interruptionObserved)
                                Thread.currentThread().interrupt();

                        throwNotificationFailure(setupFailure);
                        return;
                }

                if (notificationSessionId != null) {
                        metricsCollectorDispatcher.didOpenNotificationSession(
                                        databaseType, notificationSessionId,
                                        Duration.ofNanos(nanoTime() - lifecycleStartTime));
                }

                session = requireNonNull(session);
                Throwable callbackFailure = null;
                boolean callbackFailureIsError = false;
                InterruptedException concreteInterruptedException = null;
                boolean interruptionObserved = false;

                try {
                        if (Thread.interrupted()) {
                                interruptionObserved = true;
                        } else {
                                operation.perform(session);
                        }
                } catch (InterruptedException interruptedException) {
                        concreteInterruptedException = interruptedException;
                        interruptionObserved = true;
                        // Retain the exact exception while clearing any simultaneously pending status before cleanup.
                        Thread.interrupted();
                } catch (DatabaseException databaseException) {
                        callbackFailure = databaseException;
                } catch (Error error) {
                        callbackFailure = error;
                        callbackFailureIsError = true;
                } catch (Throwable throwable) {
                        callbackFailure = new DatabaseException(
                                        "Notification session operation failed", throwable, databaseType.dialect());
                } finally {
                        session.expire();
                }

                Throwable transportFailure = session.terminalFailure();
                Throwable primaryFailure;

                if (callbackFailureIsError) {
                        primaryFailure = appendSuppressedIfPresent(callbackFailure, transportFailure);
                } else if (transportFailure != null) {
                        primaryFailure = appendSuppressedIfPresent(transportFailure, callbackFailure);
                } else {
                        primaryFailure = callbackFailure;
                }

                // Cleanup begins flag-clear with respect to status already pending at this boundary.
                if (Thread.interrupted())
                        interruptionObserved = true;

                Throwable cleanupFailure = cleanupOpenedNotificationSession(
                                requireNonNull(connection), requireNonNull(transport), requireNonNull(initialAutoCommit),
                                session.isConnectionUncertain(), databaseType);
                primaryFailure = appendSuppressedIfPresent(primaryFailure, cleanupFailure);

                if (Thread.interrupted())
                        interruptionObserved = true;

                MetricsCollector.NotificationSessionOutcome outcome;

                if (primaryFailure != null)
                        outcome = MetricsCollector.NotificationSessionOutcome.FAILED;
                else if (interruptionObserved)
                        outcome = MetricsCollector.NotificationSessionOutcome.INTERRUPTED;
                else
                        outcome = MetricsCollector.NotificationSessionOutcome.CALLBACK_RETURNED;

                if (notificationSessionId != null) {
                        metricsCollectorDispatcher.didCloseNotificationSession(
                                        databaseType, notificationSessionId, outcome,
                                        Duration.ofNanos(nanoTime() - lifecycleStartTime),
                                        outcome == MetricsCollector.NotificationSessionOutcome.FAILED ? primaryFailure : null);
                }

                if (primaryFailure != null) {
                        if (interruptionObserved) {
                                primaryFailure = appendSuppressedIfPresent(primaryFailure, concreteInterruptedException);
                                Thread.currentThread().interrupt();
                        }

                        throwNotificationFailure(primaryFailure);
                        return;
                }

                if (interruptionObserved)
                        throw concreteInterruptedException == null ? new InterruptedException() : concreteInterruptedException;
        }

        @NonNull
        private Throwable normalizeNotificationSetupFailure(@NonNull Throwable setupCause,
                                                                                                                                                                                                                         @Nullable UnsupportedOperationException frameworkUnsupportedFailure,
                                                                                                                                                                                                                         @NonNull NotificationSetupPhase setupPhase,
                                                                                                                                                                                                                         boolean connectionFailure,
                                                                                                                                                                                                                         @NonNull DatabaseType databaseType) {
                requireNonNull(setupCause);
                requireNonNull(setupPhase);
                requireNonNull(databaseType);

                if (setupCause == frameworkUnsupportedFailure)
                        return setupCause;

                if (setupCause instanceof NotificationReceiveUnsupportedException && !connectionFailure) {
                        return new UnsupportedOperationException(
                                        "Notification listening is not supported by the current database driver connection", setupCause);
                }

                if (setupPhase == NotificationSetupPhase.OPEN_TRANSPORT
                                && !connectionFailure
                                && (setupCause instanceof SQLFeatureNotSupportedException
                                || setupCause instanceof UnsupportedOperationException
                                || setupCause instanceof AbstractMethodError)) {
                        return new UnsupportedOperationException(
                                        "Notification listening is not supported by the current database driver connection", setupCause);
                }

                return normalizeNotificationFailure("Unable to open notification session", setupCause, databaseType);
        }

        private boolean isNotificationConnectionFailure(@Nullable Connection connection,
                                                                                                                                                                                                 @NonNull Throwable throwable) {
                requireNonNull(throwable);

                if (throwable instanceof SQLException sqlException) {
                        String sqlState = sqlException.getSQLState();

                        if (sqlState != null && sqlState.startsWith("08"))
                                return true;
                }

                if (connection != null) {
                        try {
                                return connection.isClosed();
                        } catch (Throwable diagnosticFailure) {
                                if (diagnosticFailure != throwable)
                                        throwable.addSuppressed(diagnosticFailure);

                                if (diagnosticFailure instanceof SQLException diagnosticSqlException) {
                                        String sqlState = diagnosticSqlException.getSQLState();

                                        if (sqlState != null && sqlState.startsWith("08"))
                                                return true;
                                }

                                // If connection health cannot be inspected, never return the candidate to a pool as healthy.
                                return true;
                        }
                }

                return false;
        }

        @Nullable
        private Throwable cleanupFailedNotificationCandidate(@Nullable Connection connection,
                                                                                                                                                                                                                         @NonNull DatabaseType databaseType) {
                requireNonNull(databaseType);

                if (connection == null)
                        return null;

                Throwable cleanupFailure = null;

                try {
                        connection.abort(Runnable::run);
                } catch (Throwable throwable) {
                        cleanupFailure = appendSuppressed(cleanupFailure,
                                        normalizeNotificationFailure("Unable to abort failed notification-session connection",
                                                        throwable, databaseType));
                }

                try {
                        connection.close();
                } catch (Throwable throwable) {
                        cleanupFailure = appendSuppressed(cleanupFailure,
                                        normalizeNotificationFailure("Unable to close failed notification-session connection",
                                                        throwable, databaseType));
                }

                return cleanupFailure;
        }

        @Nullable
        private Throwable cleanupOpenedNotificationSession(@NonNull Connection connection,
                                                                                                                                                                                                                 @NonNull NotificationTransport transport,
                                                                                                                                                                                                                 @NonNull Boolean initialAutoCommit,
                                                                                                                                                                                                                 boolean connectionUncertain,
                                                                                                                                                                                                                 @NonNull DatabaseType databaseType) {
                requireNonNull(connection);
                requireNonNull(transport);
                requireNonNull(initialAutoCommit);
                requireNonNull(databaseType);

                Throwable cleanupFailure = null;
                boolean abortRequired = connectionUncertain;

                if (!abortRequired) {
                        try {
                                transport.unlistenAll();
                        } catch (Throwable throwable) {
                                cleanupFailure = appendSuppressed(cleanupFailure,
                                                normalizeNotificationFailure("Unable to unregister notification channels", throwable, databaseType));
                                abortRequired = true;
                        }
                }

                if (!abortRequired) {
                        try {
                                requireNonNull(transport.drain(), "Notification transport returned a null cleanup batch");

                                if (transport.isConnectionUncertain())
                                        throw new SQLException("Notification transport became uncertain during cleanup");
                        } catch (Throwable throwable) {
                                cleanupFailure = appendSuppressed(cleanupFailure,
                                                normalizeNotificationFailure("Unable to drain notification connection during cleanup",
                                                                throwable, databaseType));
                                abortRequired = true;
                        }
                }

                if (!abortRequired && !initialAutoCommit) {
                        try {
                                connection.setAutoCommit(false);
                        } catch (Throwable throwable) {
                                cleanupFailure = appendSuppressed(cleanupFailure,
                                                normalizeNotificationFailure("Unable to restore notification connection autocommit",
                                                                throwable, databaseType));
                                abortRequired = true;
                        }
                }

                if (!abortRequired) {
                        try {
                                connection.close();
                        } catch (Throwable throwable) {
                                cleanupFailure = appendSuppressed(cleanupFailure,
                                                normalizeNotificationFailure("Unable to close notification-session connection",
                                                                throwable, databaseType));
                                // Once close has been invoked, the connection handle might already have been returned to a pool.
                                // Do not attempt abort or a second close through a handle whose ownership is now unknown.
                        }
                }

                if (abortRequired) {
                        try {
                                connection.abort(Runnable::run);
                        } catch (Throwable throwable) {
                                cleanupFailure = appendSuppressed(cleanupFailure,
                                                normalizeNotificationFailure("Unable to abort uncertain notification-session connection",
                                                                throwable, databaseType));
                        }

                        try {
                                connection.close();
                        } catch (Throwable throwable) {
                                cleanupFailure = appendSuppressed(cleanupFailure,
                                                normalizeNotificationFailure("Unable to close uncertain notification-session connection",
                                                                throwable, databaseType));
                        }
                }

                return cleanupFailure;
        }

        @NonNull
        private Throwable normalizeNotificationFailure(@NonNull String message,
                                                                                                                                                                                         @NonNull Throwable throwable,
                                                                                                                                                                                         @NonNull DatabaseType databaseType) {
                requireNonNull(message);
                requireNonNull(throwable);
                requireNonNull(databaseType);

                if (throwable instanceof DatabaseException || throwable instanceof Error)
                        return throwable;

                return new DatabaseException(message, throwable, databaseType.dialect());
        }

        private static void throwNotificationFailure(@NonNull Throwable throwable) {
                requireNonNull(throwable);

                if (throwable instanceof RuntimeException runtimeException)
                        throw runtimeException;

                if (throwable instanceof Error error)
                        throw error;

                throw new DatabaseException("Notification session failed", throwable);
        }

        @Nullable
        private static Throwable appendSuppressedIfPresent(@Nullable Throwable existing,
                                                                                                                                                                                                                @Nullable Throwable additional) {
                if (additional == null)
                        return existing;

                return appendSuppressed(existing, additional);
        }

        private enum NotificationSetupPhase {
                CAPABILITY,
                ACQUIRE_CONNECTION,
                CONFIGURE_CONNECTION,
                OPEN_TRANSPORT,
                SANITIZE_UNLISTEN,
                SANITIZE_DRAIN,
                REGISTER
        }

        private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext,
                                                                                                                                                                                        @NonNull List<Object> parameters,
                                                                                                                                                                                        @NonNull DatabaseOperation databaseOperation) {
                requireNonNull(statementContext);
                requireNonNull(parameters);
                requireNonNull(databaseOperation);

                performDatabaseOperation(statementContext, parameters, null, databaseOperation);
        }

        private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext,
                                                                                                                                                                                        @NonNull List<Object> parameters,
                                                                                                                                                                                        @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                                                                                        @NonNull DatabaseOperation databaseOperation) {
                performDatabaseOperation(statementContext, parameters, preparedStatementCustomizer, databaseOperation,
                                (connection, context) -> connection.prepareStatement(context.getStatement().getSql()));
        }

        private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext,
                                                                                                                                                                                        @NonNull List<Object> parameters,
                                                                                                                                                                                        @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                                                                                        @NonNull DatabaseOperation databaseOperation,
                                                                                                                                                                                        @NonNull PreparedStatementFactory preparedStatementFactory) {
                requireNonNull(statementContext);
                requireNonNull(parameters);
                requireNonNull(databaseOperation);
                requireNonNull(preparedStatementFactory);

                performDatabaseOperation(statementContext, (preparedStatement) -> {
                        applyPreparedStatementCustomizer(statementContext, preparedStatement, preparedStatementCustomizer);
                        if (parameters.size() > 0)
                                performPreparedStatementBinding(statementContext, preparedStatement, parameters);
                }, databaseOperation, null, preparedStatementFactory);
        }

        private <T> void performPreparedStatementBinding(@NonNull StatementContext<T> statementContext,
                                                                                                                                                                                                                 @NonNull PreparedStatement preparedStatement,
                                                                                                                                                                                                                 @NonNull List<Object> parameters) {
                requireNonNull(statementContext);
                requireNonNull(preparedStatement);
                requireNonNull(parameters);

                try {
                        DefaultPreparedStatementBinder.ParameterSqlTypeResolver parameterSqlTypeResolver =
                                        new DefaultPreparedStatementBinder.ParameterSqlTypeResolver(preparedStatement);
                        PreparedStatementBinder preparedStatementBinder = preparedStatementBinderFor(statementContext);

                        for (int i = 0; i < parameters.size(); ++i) {
                                Object parameter = SecureParameterSupport.unwrapSecureAndOptionalParameter(parameters.get(i));
                                Integer parameterIndex = i + 1;

                                if (parameter != null) {
                                        if (preparedStatementBinder instanceof DefaultPreparedStatementBinder defaultPreparedStatementBinder) {
                                                defaultPreparedStatementBinder.bindParameter(statementContext, preparedStatement, parameterIndex, parameter,
                                                                parameterSqlTypeResolver);
                                        } else {
                                                preparedStatementBinder.bindParameter(statementContext, preparedStatement, parameterIndex, parameter);
                                        }
                                } else {
                                        Integer sqlType = parameterSqlTypeResolver.determineParameterSqlType(parameterIndex)
                                                        .map(DefaultPreparedStatementBinder.ParameterSqlType::getSqlType)
                                                        .orElse(Types.NULL);
                                        try {
                                                preparedStatement.setNull(parameterIndex, sqlType);
                                        } catch (SQLException | AbstractMethodError e) {
                                                if (sqlType == Types.NULL)
                                                        throw e;

                                                preparedStatement.setNull(parameterIndex, Types.NULL);
                                        }
                                }
                        }
                } catch (Exception e) {
                        throw databaseExceptionWithStatementContext(statementContext, e);
                }
        }

        private void applyPreparedStatementCustomizer(@NonNull StatementContext<?> statementContext,
                                                                                                                                                                                                        @NonNull PreparedStatement preparedStatement,
                                                                                                                                                                                                        @Nullable PreparedStatementCustomizer preparedStatementCustomizer) throws SQLException {
                requireNonNull(statementContext);
                requireNonNull(preparedStatement);

                if (preparedStatementCustomizer == null)
                        return;

                preparedStatementCustomizer.customize(statementContext, preparedStatement);
        }

        private boolean hasDefaultPreparedStatementSettings() {
                return this.queryTimeout != null || this.fetchSize != null || this.maxRows != null;
        }

        private void applyDefaultPreparedStatementSettings(@NonNull PreparedStatement preparedStatement) throws SQLException {
                requireNonNull(preparedStatement);
                applyPreparedStatementSettings(preparedStatement, this.queryTimeout, this.fetchSize, this.maxRows);
        }

        private static void applyPreparedStatementSettings(@NonNull PreparedStatement preparedStatement,
                                                                                                                                                                                                                 @Nullable Duration queryTimeout,
                                                                                                                                                                                                                 @Nullable Integer fetchSize,
                                                                                                                                                                                                                 @Nullable Integer maxRows) throws SQLException {
                requireNonNull(preparedStatement);

                if (queryTimeout != null)
                        preparedStatement.setQueryTimeout(queryTimeoutSeconds(queryTimeout));

                if (fetchSize != null)
                        preparedStatement.setFetchSize(fetchSize);

                if (maxRows != null)
                        preparedStatement.setMaxRows(maxRows);
        }

        @FunctionalInterface
        interface InternalRawConnectionOperation<R> {
                @NonNull
                Optional<R> perform(@NonNull Connection connection) throws Exception;
        }

        /**
         * Gets the database type for this database.
         * <p>
         * If {@link Builder#databaseType(DatabaseType)} was not configured and the database type has not already been detected,
         * this method may acquire a connection and inspect {@link DatabaseMetaData}.  Configure an explicit database type to avoid
         * runtime detection.
         *
         * @return the database type
         * @throws DatabaseException if automatic database type detection fails
         * @since 3.0.0
         */
        @NonNull
        public DatabaseType getDatabaseType() {
                return getDatabaseType(this.databaseTypeDetectionConnectionHolder.get());
        }

        @NonNull
        DatabaseType peekDatabaseType() {
                DatabaseType cachedDatabaseType = this.databaseType.get();
                return cachedDatabaseType == null ? DatabaseType.GENERIC : cachedDatabaseType;
        }

        @NonNull
        DatabaseDialect getDatabaseDialect() {
                return getDatabaseDialect(this.databaseTypeDetectionConnectionHolder.get());
        }

        @NonNull
        DatabaseDialect getDatabaseDialect(@Nullable Connection connection) {
                DatabaseDialect cachedDatabaseDialect = this.databaseDialect.get();

                if (cachedDatabaseDialect != null)
                        return cachedDatabaseDialect;

                DatabaseDialect detectedDatabaseDialect = getDatabaseType(connection).dialect();

                if (this.databaseDialect.compareAndSet(null, detectedDatabaseDialect))
                        return detectedDatabaseDialect;

                return requireNonNull(this.databaseDialect.get());
        }

        private void warmDatabaseTypeCacheForMetricsIfNeeded(@NonNull StatementContext<?> statementContext) {
                requireNonNull(statementContext);

                if (!getMetricsCollectorDispatcher().isEnabled())
                        return;

                // Trigger lazy database-type detection while the statement's JDBC connection is active so later
                // transaction-scope metrics can report an accurate db.system.name without opening a second metadata connection.
                try {
                        statementContext.getDatabaseType();
                } catch (Throwable t) {
                        this.logger.log(Level.FINE, "Unable to warm database type cache for metrics", t);
                }
        }

        private void dispatchWithDatabaseTypeDetectionConnection(@NonNull Connection connection,
                                                                                                                                                                                                                                         @NonNull Runnable operation) {
                requireNonNull(connection);
                requireNonNull(operation);

                Connection previousDatabaseTypeDetectionConnection = this.databaseTypeDetectionConnectionHolder.get();
                this.databaseTypeDetectionConnectionHolder.set(connection);

                try {
                        operation.run();
                } finally {
                        if (previousDatabaseTypeDetectionConnection == null)
                                this.databaseTypeDetectionConnectionHolder.remove();
                        else
                                this.databaseTypeDetectionConnectionHolder.set(previousDatabaseTypeDetectionConnection);
                }
        }

        @NonNull
        private DatabaseType getDatabaseType(@Nullable Connection connection) {
                DatabaseType cachedDatabaseType = this.databaseType.get();

                if (cachedDatabaseType != null)
                        return cachedDatabaseType;

                DatabaseType detectedDatabaseType;

                try {
                        detectedDatabaseType = connection == null
                                        ? DatabaseType.fromDataSource(getDataSource())
                                        : DatabaseType.fromConnection(connection);
                } catch (DatabaseException e) {
                        throw new DatabaseException(format(
                                        "Unable to determine database type automatically. Configure %s.%s(%s) explicitly to avoid runtime detection.",
                                        Builder.class.getSimpleName(), "databaseType", DatabaseType.class.getSimpleName()), e);
                }

                if (this.databaseType.compareAndSet(null, detectedDatabaseType))
                        return detectedDatabaseType;

                return this.databaseType.get();
        }

        /**
         * @since 3.0.0
         */
        @NonNull
        public ZoneId getTimeZone() {
                return this.timeZone;
        }

        /**
         * How should Pyranid bind {@link java.time.Instant} and {@link java.time.OffsetDateTime} parameters when JDBC
         * parameter metadata cannot identify whether the target is {@code TIMESTAMP} or {@code TIMESTAMP WITH TIME ZONE}?
         *
         * @return behavior to use when timestamp target metadata is unavailable or non-identifying
         * @since 4.2.0
         */
        @NonNull
        public AmbiguousTimestampBindingStrategy getAmbiguousTimestampBindingStrategy() {
                return this.ambiguousTimestampBindingStrategy;
        }

        /**
         * Gets the configured redactor used for non-secure parameters in diagnostics.
         *
         * @return the configured parameter redactor
         * @since 4.4.0
         */
        @NonNull
        public ParameterRedactor getParameterRedactor() {
                return this.parameterRedactor;
        }

        /**
         * Useful for single-shot "utility" calls that operate outside of normal query operations, e.g. pulling DB metadata.
         * <p>
         * Example: {@link #readDatabaseMetaData(DatabaseMetaDataReader)}.
         */
        @NonNull
        <R> Optional<R> performRawConnectionOperation(@NonNull InternalRawConnectionOperation<R> rawConnectionOperation,
                                                                                                                                                                                                                                        @NonNull Boolean shouldParticipateInExistingTransactionIfPossible) {
                requireNonNull(rawConnectionOperation);
                requireNonNull(shouldParticipateInExistingTransactionIfPossible);

                if (shouldParticipateInExistingTransactionIfPossible) {
                        Optional<Transaction> transaction = currentTransactionForDatabaseOperation();
                        ReentrantLock connectionLock = transaction.isPresent() ? transaction.get().getConnectionLock() : null;
                        // Try to participate in txn if it's available
                        Connection connection = null;
                        Throwable thrown = null;
                        boolean connectionLockAcquired = false;

                        try {
                                if (connectionLock != null) {
                                        lockInterruptibly(connectionLock, "use the transaction connection for a raw connection operation");
                                        connectionLockAcquired = true;
                                }

                                if (transaction.isPresent()) {
                                        connection = transaction.get().getConnection();
                                } else {
                                        try {
                                                connection = getDataSource().getConnection();
                                        } catch (SQLException e) {
                                                throw new DatabaseException("Unable to acquire database connection", e);
                                        }
                                }

                                return rawConnectionOperation.perform(connection);
                        } catch (DatabaseException e) {
                                thrown = e;
                                throw e;
                        } catch (Error e) {
                                thrown = e;
                                throw e;
                        } catch (Exception e) {
                                DatabaseException wrapped = databaseExceptionWithRawConnectionContext(connection, e);
                                thrown = wrapped;
                                throw wrapped;
                        } finally {
                                Throwable cleanupFailure = null;

                                try {
                                        // If this was a single-shot operation (not in a transaction), close the connection
                                        if (connection != null && !transaction.isPresent()) {
                                                try {
                                                        closeConnection(connection);
                                                } catch (Throwable cleanupException) {
                                                        cleanupFailure = cleanupException;
                                                }
                                        }
                                } finally {
                                        if (connectionLockAcquired)
                                                connectionLock.unlock();

                                        if (cleanupFailure != null) {
                                                if (thrown != null) {
                                                        thrown.addSuppressed(cleanupFailure);
                                                } else if (cleanupFailure instanceof RuntimeException) {
                                                        throw (RuntimeException) cleanupFailure;
                                                } else if (cleanupFailure instanceof Error) {
                                                        throw (Error) cleanupFailure;
                                                } else {
                                                        throw new RuntimeException(cleanupFailure);
                                                }
                                        }
                                }
                        }
                } else {
                        boolean acquiredConnection = false;
                        Connection connection = null;
                        Throwable thrown = null;

                        // Always get a fresh connection no matter what and close it afterwards
                        try {
                                connection = getDataSource().getConnection();
                                acquiredConnection = true;
                                return rawConnectionOperation.perform(connection);
                        } catch (DatabaseException e) {
                                thrown = e;
                                throw e;
                        } catch (Error e) {
                                thrown = e;
                                throw e;
                        } catch (Exception e) {
                                DatabaseException wrapped = acquiredConnection
                                                ? databaseExceptionWithRawConnectionContext(connection, e)
                                                : new DatabaseException("Unable to acquire database connection", e);
                                thrown = wrapped;
                                throw wrapped;
                        } finally {
                                if (connection != null) {
                                        try {
                                                closeConnection(connection);
                                        } catch (Throwable cleanupException) {
                                                if (thrown != null) {
                                                        thrown.addSuppressed(cleanupException);
                                                } else if (cleanupException instanceof RuntimeException) {
                                                        throw (RuntimeException) cleanupException;
                                                } else if (cleanupException instanceof Error) {
                                                        throw (Error) cleanupException;
                                                } else {
                                                        throw new RuntimeException(cleanupException);
                                                }
                                        }
                                }
                        }
                }
        }

        private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext,
                                                                                                                                                                                        @NonNull PreparedStatementBindingOperation preparedStatementBindingOperation,
                                                                                                                                                                                        @NonNull DatabaseOperation databaseOperation) {
                performDatabaseOperation(statementContext, preparedStatementBindingOperation, databaseOperation, null);
        }

        private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext,
                                                                                                                                                                                        @NonNull PreparedStatementBindingOperation preparedStatementBindingOperation,
                                                                                                                                                                                        @NonNull DatabaseOperation databaseOperation,
                                                                                                                                                                                        @Nullable Integer batchSize) {
                performDatabaseOperation(statementContext, preparedStatementBindingOperation, databaseOperation, batchSize,
                                (connection, context) -> connection.prepareStatement(context.getStatement().getSql()));
        }

        private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext,
                                                                                                                                                                                        @NonNull PreparedStatementBindingOperation preparedStatementBindingOperation,
                                                                                                                                                                                        @NonNull DatabaseOperation databaseOperation,
                                                                                                                                                                                        @Nullable Integer batchSize,
                                                                                                                                                                                        @NonNull PreparedStatementFactory preparedStatementFactory) {
                requireNonNull(statementContext);
                requireNonNull(preparedStatementBindingOperation);
                requireNonNull(databaseOperation);
                requireNonNull(preparedStatementFactory);

                long startTime = nanoTime();
                Duration connectionAcquisitionDuration = null;
                Duration preparationDuration = null;
                Duration executionDuration = null;
                Duration resultSetMappingDuration = null;
                StatementResult statementResult = StatementResult.empty();
                Exception exception = null;
                Throwable thrown = null;
                Connection connection = null;
                long connectionHeldStartTime = 0L;
                Optional<Transaction> transaction = currentTransactionForDatabaseOperation();
                ReentrantLock connectionLock = transaction.isPresent() ? transaction.get().getConnectionLock() : null;
                boolean connectionLockAcquired = false;
                Connection previousDatabaseTypeDetectionConnection = null;
                boolean databaseTypeDetectionConnectionInstalled = false;

                try {
                        if (connectionLock != null) {
                                lockInterruptibly(connectionLock, "execute a statement on the transaction connection");
                                connectionLockAcquired = true;
                        }

                        boolean alreadyHasConnection = transaction.isPresent() && transaction.get().hasConnection();
                        if (transaction.isPresent()) {
                                connection = transaction.get().getConnection();
                        } else {
                                getMetricsCollectorDispatcher().willAcquireStatementConnection(statementContext);
                                try {
                                        connection = getDataSource().getConnection();
                                } catch (SQLException e) {
                                        DatabaseException wrapped = new DatabaseException("Unable to acquire database connection", e);
                                        connectionAcquisitionDuration = Duration.ofNanos(nanoTime() - startTime);
                                        getMetricsCollectorDispatcher().didFailToAcquireStatementConnection(statementContext, peekDatabaseType(),
                                                        connectionAcquisitionDuration, wrapped);
                                        throw wrapped;
                                } catch (RuntimeException e) {
                                        connectionAcquisitionDuration = Duration.ofNanos(nanoTime() - startTime);
                                        getMetricsCollectorDispatcher().didFailToAcquireStatementConnection(statementContext, peekDatabaseType(),
                                                        connectionAcquisitionDuration, e);
                                        throw e;
                                }
                        }

                        connectionAcquisitionDuration = alreadyHasConnection ? null : Duration.ofNanos(nanoTime() - startTime);
                        previousDatabaseTypeDetectionConnection = this.databaseTypeDetectionConnectionHolder.get();
                        this.databaseTypeDetectionConnectionHolder.set(connection);
                        databaseTypeDetectionConnectionInstalled = true;

                        if (!transaction.isPresent()) {
                                connectionHeldStartTime = nanoTime();
                                Duration acquiredDuration = connectionAcquisitionDuration;
                                MetricsCollectorDispatcher metricsCollectorDispatcher = getMetricsCollectorDispatcher();
                                if (metricsCollectorDispatcher.isEnabled())
                                        dispatchWithDatabaseTypeDetectionConnection(connection, () ->
                                                        metricsCollectorDispatcher.didAcquireStatementConnection(statementContext, acquiredDuration));
                        }
                        startTime = nanoTime();

                        try (PreparedStatement preparedStatement = preparedStatementFactory.prepare(connection, statementContext)) {
                                preparedStatementBindingOperation.perform(preparedStatement);
                                preparationDuration = Duration.ofNanos(nanoTime() - startTime);

                                getMetricsCollectorDispatcher().willExecuteStatement(statementContext);
                                DatabaseOperationResult databaseOperationResult = databaseOperation.perform(preparedStatement);
                                executionDuration = databaseOperationResult.getExecutionDuration().orElse(null);
                                resultSetMappingDuration = databaseOperationResult.getResultSetMappingDuration().orElse(null);
                                statementResult = databaseOperationResult.getStatementResult();
                                warmDatabaseTypeCacheForMetricsIfNeeded(statementContext);
                        }
                } catch (DatabaseException e) {
                        DatabaseException wrapped = databaseExceptionWithStatementContext(statementContext, e);
                        exception = wrapped;
                        thrown = wrapped;
                        throw wrapped;
                } catch (Error e) {
                        exception = databaseExceptionWithStatementContext(statementContext, e);
                        thrown = e;
                        throw e;
                } catch (Exception e) {
                        DatabaseException wrapped = databaseExceptionWithStatementContext(statementContext, e);
                        // Store the wrapped (scrubbed) form in the StatementLog, consistent with the DatabaseException
                        // and Error paths - the raw driver exception remains available via the wrapped cause
                        exception = wrapped;
                        thrown = wrapped;
                        throw wrapped;
                } finally {
                        Throwable cleanupFailure = null;

                        if (databaseTypeDetectionConnectionInstalled) {
                                if (previousDatabaseTypeDetectionConnection == null)
                                        this.databaseTypeDetectionConnectionHolder.remove();
                                else
                                        this.databaseTypeDetectionConnectionHolder.set(previousDatabaseTypeDetectionConnection);
                        }

                        try {
                                cleanupFailure = closeStatementContextResources(statementContext, cleanupFailure);

                                // If this was a single-shot operation (not in a transaction), close the connection
                                if (connection != null && !transaction.isPresent()) {
                                        Duration heldDuration = Duration.ofNanos(nanoTime() - connectionHeldStartTime);
                                        try {
                                                closeConnection(connection);
                                                getMetricsCollectorDispatcher().didReleaseStatementConnection(statementContext, heldDuration);
                                        } catch (Throwable cleanupException) {
                                                getMetricsCollectorDispatcher().didFailToReleaseStatementConnection(statementContext, heldDuration, cleanupException);
                                                if (cleanupFailure == null)
                                                        cleanupFailure = cleanupException;
                                                else
                                                        cleanupFailure.addSuppressed(cleanupException);
                                        }
                                }
                        } finally {
                                if (connectionLockAcquired)
                                        connectionLock.unlock();

                                StatementLog statementLog =
                                                StatementLog.withStatementContext(statementContext)
                                                                .connectionAcquisitionDuration(connectionAcquisitionDuration)
                                                                .preparationDuration(preparationDuration)
                                                                .executionDuration(executionDuration)
                                                                .resultSetMappingDuration(resultSetMappingDuration)
                                                                .batchSize(batchSize)
                                                                .exception(exception)
                                                                .build();

                                if (thrown == null && exception == null) {
                                        getMetricsCollectorDispatcher().didExecuteStatement(statementContext, statementLog, statementResult);
                                } else {
                                        Throwable statementThrowable = thrown == null ? exception : thrown;
                                        getMetricsCollectorDispatcher().didFailToExecuteStatement(statementContext, statementLog, peekDatabaseType(),
                                                        requireNonNull(statementThrowable));
                                }

                                try {
                                        getStatementLogger().log(statementLog);
                                } catch (Throwable cleanupException) {
                                        if (cleanupFailure == null)
                                                cleanupFailure = cleanupException;
                                        else
                                                cleanupFailure.addSuppressed(cleanupException);
                                }
                        }

                        if (cleanupFailure != null) {
                                if (thrown != null) {
                                        thrown.addSuppressed(cleanupFailure);
                                } else if (cleanupFailure instanceof RuntimeException) {
                                        throw (RuntimeException) cleanupFailure;
                                } else if (cleanupFailure instanceof Error) {
                                        throw (Error) cleanupFailure;
                                } else {
                                        throw new RuntimeException(cleanupFailure);
                                }
                        }
                }
        }

        @NonNull
        DataSource getDataSource() {
                return this.dataSource;
        }

        @NonNull
        InstanceProvider getInstanceProvider() {
                return this.instanceProvider;
        }

        @NonNull
        private PreparedStatementBinder getPreparedStatementBinder() {
                return this.preparedStatementBinder;
        }

        /**
         * Resolves the effective {@link ResultSetMapper} for a statement: the per-query override carried on
         * the {@link StatementContext} when present, otherwise the database-wide instance.
         */
        @NonNull
        private ResultSetMapper resultSetMapperFor(@NonNull StatementContext<?> statementContext) {
                requireNonNull(statementContext);
                ResultSetMapper resultSetMapperOverride = statementContext.getResultSetMapperOverride();
                return resultSetMapperOverride != null ? resultSetMapperOverride : getResultSetMapper();
        }

        /**
         * Resolves the effective {@link PreparedStatementBinder} for a statement: the per-query override
         * carried on the {@link StatementContext} when present, otherwise the database-wide instance.
         */
        @NonNull
        private PreparedStatementBinder preparedStatementBinderFor(@NonNull StatementContext<?> statementContext) {
                requireNonNull(statementContext);
                PreparedStatementBinder preparedStatementBinderOverride = statementContext.getPreparedStatementBinderOverride();
                return preparedStatementBinderOverride != null ? preparedStatementBinderOverride : getPreparedStatementBinder();
        }

        @NonNull
        ResultSetMapper getResultSetMapper() {
                return this.resultSetMapper;
        }

        @NonNull
        private StatementLogger getStatementLogger() {
                return this.statementLogger;
        }

        @NonNull
        public MetricsCollector getMetricsCollector() {
                return getMetricsCollectorDispatcher().getMetricsCollector();
        }

        @NonNull
        MetricsCollectorDispatcher getMetricsCollectorDispatcher() {
                return this.metricsCollectorDispatcher;
        }

        @NonNull
        private DatabaseOperationSupportStatus getExecuteLargeBatchSupported() {
                return this.executeLargeBatchSupported;
        }

        private void setExecuteLargeBatchSupported(@NonNull DatabaseOperationSupportStatus executeLargeBatchSupported) {
                requireNonNull(executeLargeBatchSupported);
                this.executeLargeBatchSupported = executeLargeBatchSupported;
        }

        @NonNull
        private DatabaseOperationSupportStatus getExecuteLargeUpdateSupported() {
                return this.executeLargeUpdateSupported;
        }

        private void setExecuteLargeUpdateSupported(@NonNull DatabaseOperationSupportStatus executeLargeUpdateSupported) {
                requireNonNull(executeLargeUpdateSupported);
                this.executeLargeUpdateSupported = executeLargeUpdateSupported;
        }

        @NonNull
        Object generateId() {
                // "Unique" keys
                return this.defaultIdGenerator.incrementAndGet();
        }

        @FunctionalInterface
        private interface DatabaseOperation {
                @NonNull
                DatabaseOperationResult perform(@NonNull PreparedStatement preparedStatement) throws Exception;
        }

        @FunctionalInterface
        private interface PreparedStatementFactory {
                @NonNull
                PreparedStatement prepare(@NonNull Connection connection,
                                                                                                                        @NonNull StatementContext<?> statementContext) throws SQLException;
        }

        @FunctionalInterface
        private interface PreparedStatementBindingOperation {
                void perform(@NonNull PreparedStatement preparedStatement) throws Exception;
        }

        @NotThreadSafe
        private static final class StreamingResultSet<T> implements java.util.Iterator<T>, AutoCloseable {
                private final Database database;
                private final StatementContext<T> statementContext;
                private final List<Object> parameters;
                @Nullable
                private final PreparedStatementCustomizer preparedStatementCustomizer;
                @NonNull
                private final Optional<Transaction> transaction;
                @Nullable
                private final ReentrantLock connectionLock;
                @Nullable
                private Connection connection;
                @Nullable
                private PreparedStatement preparedStatement;
                @Nullable
                private ResultSet resultSet;
                private boolean closed;
                private boolean hasNextEvaluated;
                private boolean hasNext;
                @Nullable
                private Duration connectionAcquisitionDuration;
                @Nullable
                private Duration preparationDuration;
                @Nullable
                private Duration executionDuration;
                private long resultSetMappingNanos;
                @Nullable
                private Exception exception;
                @Nullable
                private Throwable thrown;
                private long rowsConsumed;
                private long openStartTime;
                private boolean exhausted;
                private boolean openFailed;
                private boolean terminalMetricsEmitted;
                @Nullable
                private Throwable callbackThrowable;
                @Nullable
                private Throwable iterationThrowable;
                @Nullable
                private Throwable cleanupFailure;
                private long connectionHeldStartTime;
                @NonNull
                private DatabaseDialect databaseStreamDialect = GenericDialect.INSTANCE;
                @NonNull
                private DatabaseStreamState databaseStreamState = DatabaseStreamState.none();
                private boolean connectionLockAcquired;
                private final boolean queryFetchSizeConfigured;

                private StreamingResultSet(@NonNull Database database,
                                                                                                                         @NonNull StatementContext<T> statementContext,
                                                                                                                         @NonNull List<Object> parameters,
                                                                                                                         @Nullable PreparedStatementCustomizer preparedStatementCustomizer,
                                                                                                                         boolean queryFetchSizeConfigured) {
                        this.database = requireNonNull(database);
                        this.statementContext = requireNonNull(statementContext);
                        this.parameters = requireNonNull(parameters);
                        this.preparedStatementCustomizer = preparedStatementCustomizer;
                        this.transaction = database.currentTransactionForDatabaseOperation();
                        this.connectionLock = this.transaction.isPresent() ? this.transaction.get().getConnectionLock() : null;
                        this.queryFetchSizeConfigured = queryFetchSizeConfigured;

                        open();
                }

                private void open() {
                        long startTime = nanoTime();
                        this.openStartTime = startTime;
                        this.database.getMetricsCollectorDispatcher().willOpenStream(this.statementContext);
                        Connection previousDatabaseTypeDetectionConnection = null;
                        boolean databaseTypeDetectionConnectionInstalled = false;

                        try {
                                if (this.connectionLock != null) {
                                        lockInterruptibly(this.connectionLock, "open a stream on the transaction connection");
                                        this.connectionLockAcquired = true;
                                }

                                boolean alreadyHasConnection = this.transaction.isPresent() && this.transaction.get().hasConnection();
                                if (this.transaction.isPresent()) {
                                        this.connection = this.transaction.get().getConnection();
                                } else {
                                        this.database.getMetricsCollectorDispatcher().willAcquireStatementConnection(this.statementContext);
                                        try {
                                                this.connection = this.database.getDataSource().getConnection();
                                        } catch (SQLException e) {
                                                DatabaseException wrapped = new DatabaseException("Unable to acquire database connection", e);
                                                this.connectionAcquisitionDuration = Duration.ofNanos(nanoTime() - startTime);
                                                this.database.getMetricsCollectorDispatcher().didFailToAcquireStatementConnection(this.statementContext,
                                                                this.database.peekDatabaseType(), this.connectionAcquisitionDuration, wrapped);
                                                throw wrapped;
                                        } catch (RuntimeException e) {
                                                this.connectionAcquisitionDuration = Duration.ofNanos(nanoTime() - startTime);
                                                this.database.getMetricsCollectorDispatcher().didFailToAcquireStatementConnection(this.statementContext,
                                                                this.database.peekDatabaseType(), this.connectionAcquisitionDuration, e);
                                                throw e;
                                        }
                                }
                                this.connectionAcquisitionDuration = alreadyHasConnection ? null : Duration.ofNanos(nanoTime() - startTime);
                                previousDatabaseTypeDetectionConnection = this.database.databaseTypeDetectionConnectionHolder.get();
                                this.database.databaseTypeDetectionConnectionHolder.set(this.connection);
                                databaseTypeDetectionConnectionInstalled = true;

                                if (this.transaction.isEmpty()) {
                                        this.connectionHeldStartTime = nanoTime();
                                        MetricsCollectorDispatcher metricsCollectorDispatcher = this.database.getMetricsCollectorDispatcher();
                                        if (metricsCollectorDispatcher.isEnabled())
                                                this.database.dispatchWithDatabaseTypeDetectionConnection(requireNonNull(this.connection), () ->
                                                                metricsCollectorDispatcher.didAcquireStatementConnection(this.statementContext, this.connectionAcquisitionDuration));
                                }
                                startTime = nanoTime();

                                this.databaseStreamDialect = databaseDialectForStreamingConnection();
                                this.databaseStreamState = this.databaseStreamDialect.configureStreamingConnection(requireNonNull(this.connection), this.transaction.isPresent());

                                this.preparedStatement = this.databaseStreamDialect.prepareStreamingStatement(requireNonNull(this.connection), this.statementContext);
                                this.database.applyPreparedStatementCustomizer(this.statementContext, this.preparedStatement, this.preparedStatementCustomizer);
                                this.databaseStreamDialect.configureStreamingPreparedStatement(this.preparedStatement, this.databaseStreamState,
                                                this.transaction.isPresent(), this.queryFetchSizeConfigured);
                                if (this.parameters.size() > 0)
                                        this.database.performPreparedStatementBinding(this.statementContext, this.preparedStatement, this.parameters);
                                this.preparationDuration = Duration.ofNanos(nanoTime() - startTime);

                                startTime = nanoTime();
                                this.resultSet = this.preparedStatement.executeQuery();
                                this.executionDuration = Duration.ofNanos(nanoTime() - startTime);
                                this.database.warmDatabaseTypeCacheForMetricsIfNeeded(this.statementContext);
                                this.database.getMetricsCollectorDispatcher().didOpenStream(this.statementContext, Duration.ofNanos(nanoTime() - this.openStartTime));
                        } catch (DatabaseException e) {
                                DatabaseException wrapped = databaseExceptionWithStatementContext(this.statementContext, e);
                                this.exception = wrapped;
                                this.thrown = wrapped;
                                this.openFailed = true;
                                this.database.getMetricsCollectorDispatcher().didFailToOpenStream(this.statementContext, this.database.peekDatabaseType(),
                                                Duration.ofNanos(nanoTime() - this.openStartTime), wrapped);
                                close();
                                throw wrapped;
                        } catch (Exception e) {
                                DatabaseException wrapped = databaseExceptionWithStatementContext(this.statementContext, e);
                                this.exception = wrapped;
                                this.thrown = wrapped;
                                this.openFailed = true;
                                this.database.getMetricsCollectorDispatcher().didFailToOpenStream(this.statementContext, this.database.peekDatabaseType(),
                                                Duration.ofNanos(nanoTime() - this.openStartTime), wrapped);
                                close();
                                throw wrapped;
                        } catch (Error e) {
                                this.exception = databaseExceptionWithStatementContext(this.statementContext, e);
                                this.thrown = e;
                                this.openFailed = true;
                                this.database.getMetricsCollectorDispatcher().didFailToOpenStream(this.statementContext, this.database.peekDatabaseType(),
                                                Duration.ofNanos(nanoTime() - this.openStartTime), e);
                                close();
                                throw e;
                        } finally {
                                if (databaseTypeDetectionConnectionInstalled) {
                                        if (previousDatabaseTypeDetectionConnection == null)
                                                this.database.databaseTypeDetectionConnectionHolder.remove();
                                        else
                                                this.database.databaseTypeDetectionConnectionHolder.set(previousDatabaseTypeDetectionConnection);
                                }
                        }
                }

                @NonNull
                private DatabaseDialect databaseDialectForStreamingConnection() {
                        try {
                                return this.database.getDatabaseDialect(requireNonNull(this.connection));
                        } catch (DatabaseException e) {
                                return this.database.peekDatabaseType().dialect();
                        }
                }

                @Override
                public boolean hasNext() {
                        if (this.closed)
                                return false;

                        if (!this.hasNextEvaluated) {
                                try {
                                        this.hasNext = this.resultSet != null && this.resultSet.next();
                                        this.hasNextEvaluated = true;
                                        if (!this.hasNext) {
                                                this.exhausted = true;
                                                close();
                                        }
                                } catch (SQLException e) {
                                        DatabaseException wrapped = databaseExceptionWithStatementContext(this.statementContext, e);
                                        this.exception = wrapped;
                                        this.thrown = wrapped;
                                        this.iterationThrowable = wrapped;
                                        close();
                                        throw wrapped;
                                } catch (RuntimeException e) {
                                        this.exception = e;
                                        this.thrown = e;
                                        this.iterationThrowable = e;
                                        close();
                                        throw e;
                                } catch (Error e) {
                                        this.exception = databaseExceptionWithStatementContext(this.statementContext, e);
                                        this.thrown = e;
                                        this.iterationThrowable = e;
                                        close();
                                        throw e;
                                }
                        }

                        return this.hasNext;
                }

                @Override
                public T next() {
                        if (!hasNext())
                                throw new java.util.NoSuchElementException();

                        this.hasNextEvaluated = false;
                        long startTime = nanoTime();
                        Connection previousDatabaseTypeDetectionConnection = this.database.databaseTypeDetectionConnectionHolder.get();
                        this.database.databaseTypeDetectionConnectionHolder.set(requireNonNull(this.connection));

                        try {
                                T value = this.database.resultSetMapperFor(this.statementContext)
                                                .map(this.statementContext, requireNonNull(this.resultSet), this.statementContext.getResultSetRowType().get(), this.database.getInstanceProvider())
                                                .orElse(null);

                                this.resultSetMappingNanos += nanoTime() - startTime;
                                this.rowsConsumed++;
                                return value;
                        } catch (SQLException e) {
                                DatabaseException wrapped = databaseExceptionWithStatementContext(this.statementContext,
                                                format("Unable to map JDBC %s row to %s", ResultSet.class.getSimpleName(), this.statementContext.getResultSetRowType().get()), e);
                                this.exception = wrapped;
                                this.thrown = wrapped;
                                this.iterationThrowable = wrapped;
                                close();
                                throw wrapped;
                        } catch (DatabaseException e) {
                                DatabaseException wrapped = databaseExceptionWithStatementContext(this.statementContext, e);
                                this.exception = wrapped;
                                this.thrown = wrapped;
                                this.iterationThrowable = wrapped;
                                close();
                                throw wrapped;
                        } catch (RuntimeException e) {
                                this.exception = e;
                                this.thrown = e;
                                this.iterationThrowable = e;
                                close();
                                throw e;
                        } catch (Error e) {
                                this.exception = databaseExceptionWithStatementContext(this.statementContext,
                                                format("Unable to map JDBC %s row to %s", ResultSet.class.getSimpleName(), this.statementContext.getResultSetRowType().get()), e);
                                this.thrown = e;
                                this.iterationThrowable = e;
                                close();
                                throw e;
                        } finally {
                                if (previousDatabaseTypeDetectionConnection == null)
                                        this.database.databaseTypeDetectionConnectionHolder.remove();
                                else
                                        this.database.databaseTypeDetectionConnectionHolder.set(previousDatabaseTypeDetectionConnection);
                        }
                }

                private void callbackFailed(@NonNull Throwable throwable) {
                        requireNonNull(throwable);
                        this.callbackThrowable = throwable;
                }

                @Override
                public void close() {
                        if (this.closed)
                                return;

                        if (this.connectionLockAcquired && this.connectionLock != null && !this.connectionLock.isHeldByCurrentThread())
                                throw new DatabaseException("Transactional streams must be closed by the thread that opened them");

                        this.closed = true;
                        Throwable cleanupFailure = null;

                        try {
                                cleanupFailure = closeStatementContextResources(this.statementContext, cleanupFailure);

                                if (this.resultSet != null) {
                                        try {
                                                this.resultSet.close();
                                        } catch (Throwable cleanupException) {
                                                cleanupFailure = cleanupFailure == null ? cleanupException : addSuppressed(cleanupFailure, cleanupException);
                                        }
                                }

                                if (this.preparedStatement != null) {
                                        try {
                                                this.preparedStatement.close();
                                        } catch (Throwable cleanupException) {
                                                cleanupFailure = cleanupFailure == null ? cleanupException : addSuppressed(cleanupFailure, cleanupException);
                                        }
                                }

                                cleanupFailure = completeDialectStreamingConnectionIfNeeded(cleanupFailure);

                                if (this.connection != null && this.transaction.isEmpty()) {
                                        Duration heldDuration = this.connectionHeldStartTime == 0L
                                                        ? Duration.ZERO
                                                        : Duration.ofNanos(nanoTime() - this.connectionHeldStartTime);
                                        try {
                                                this.database.closeConnection(this.connection);
                                                this.database.getMetricsCollectorDispatcher().didReleaseStatementConnection(this.statementContext, heldDuration);
                                        } catch (Throwable cleanupException) {
                                                this.database.getMetricsCollectorDispatcher().didFailToReleaseStatementConnection(this.statementContext, heldDuration, cleanupException);
                                                cleanupFailure = cleanupFailure == null ? cleanupException : addSuppressed(cleanupFailure, cleanupException);
                                        }
                                }
                        } finally {
                                if (this.connectionLockAcquired) {
                                        this.connectionLock.unlock();
                                        this.connectionLockAcquired = false;
                                }

                                Duration mappingDuration = this.resultSetMappingNanos == 0L ? null : Duration.ofNanos(this.resultSetMappingNanos);

                                StatementLog statementLog =
                                                StatementLog.withStatementContext(this.statementContext)
                                                                .connectionAcquisitionDuration(this.connectionAcquisitionDuration)
                                                                .preparationDuration(this.preparationDuration)
                                                                .executionDuration(this.executionDuration)
                                                                .resultSetMappingDuration(mappingDuration)
                                                                .exception(this.exception)
                                                                .build();

                                if (this.thrown == null && this.exception == null) {
                                        this.database.getMetricsCollectorDispatcher().didExecuteStatement(this.statementContext, statementLog, StatementResult.empty());
                                } else {
                                        Throwable statementThrowable = this.thrown == null ? this.exception : this.thrown;
                                        this.database.getMetricsCollectorDispatcher().didFailToExecuteStatement(this.statementContext, statementLog,
                                                        this.database.peekDatabaseType(), requireNonNull(statementThrowable));
                                }

                                try {
                                        this.database.getStatementLogger().log(statementLog);
                                } catch (Throwable cleanupException) {
                                        cleanupFailure = cleanupFailure == null ? cleanupException : addSuppressed(cleanupFailure, cleanupException);
                                }
                        }

                        this.cleanupFailure = cleanupFailure;

                        if (cleanupFailure != null) {
                                if (this.thrown != null) {
                                        this.thrown.addSuppressed(cleanupFailure);
                                } else if (cleanupFailure instanceof RuntimeException) {
                                        throw (RuntimeException) cleanupFailure;
                                } else if (cleanupFailure instanceof Error) {
                                        throw (Error) cleanupFailure;
                                } else {
                                        throw new RuntimeException(cleanupFailure);
                                }
                        }
                }

                @Nullable
                private Throwable completeDialectStreamingConnectionIfNeeded(@Nullable Throwable cleanupFailure) {
                        if (this.connection == null)
                                return cleanupFailure;

                        boolean streamSucceeded = this.thrown == null && this.exception == null && this.callbackThrowable == null && !this.openFailed;
                        cleanupFailure = this.databaseStreamDialect.completeStreamingConnection(requireNonNull(this.connection),
                                        this.databaseStreamState, streamSucceeded, cleanupFailure);

                        this.databaseStreamDialect = GenericDialect.INSTANCE;
                        this.databaseStreamState = DatabaseStreamState.none();
                        return cleanupFailure;
                }

                private void emitTerminalMetrics() {
                        if (this.terminalMetricsEmitted)
                                return;

                        this.terminalMetricsEmitted = true;

                        if (this.openFailed)
                                return;

                        MetricsCollector.StreamTerminalOutcome outcome;
                        Throwable throwable;

                        if (this.iterationThrowable != null) {
                                outcome = MetricsCollector.StreamTerminalOutcome.ITERATION_FAILURE;
                                throwable = this.iterationThrowable;
                        } else if (this.callbackThrowable != null) {
                                outcome = MetricsCollector.StreamTerminalOutcome.CALLBACK_FAILURE;
                                throwable = this.callbackThrowable;
                        } else if (this.exhausted) {
                                outcome = MetricsCollector.StreamTerminalOutcome.COMPLETED_NORMALLY;
                                throwable = this.cleanupFailure;
                        } else {
                                outcome = MetricsCollector.StreamTerminalOutcome.EARLY_CLOSE;
                                throwable = this.cleanupFailure;
                        }

                        this.database.getMetricsCollectorDispatcher().didCloseStream(this.statementContext, outcome, this.rowsConsumed,
                                        Duration.ofNanos(nanoTime() - this.openStartTime), throwable);
                }

                @NonNull
                private static Throwable addSuppressed(@NonNull Throwable existing,
                                                                                                                                                                         @NonNull Throwable additional) {
                        existing.addSuppressed(additional);
                        return existing;
                }
        }

        /**
         * Builder used to construct instances of {@link Database}.
         * <p>
         * This class is intended for use by a single thread.
         *
         * @author <a href="https://www.revetkn.com">Mark Allen</a>
         * @since 1.0.0
         */
        @NotThreadSafe
        public static class Builder {
                @NonNull
                private final DataSource dataSource;
                @Nullable
                private DatabaseType databaseType;
                @Nullable
                private ZoneId timeZone;
                @Nullable
                private AmbiguousTimestampBindingStrategy ambiguousTimestampBindingStrategy;
                @Nullable
                private InstanceProvider instanceProvider;
                @Nullable
                private PreparedStatementBinder preparedStatementBinder;
                @Nullable
                private ResultSetMapper resultSetMapper;
                @Nullable
                private StatementLogger statementLogger;
                @Nullable
                private ParameterRedactor parameterRedactor;
                @Nullable
                private MetricsCollector metricsCollector;
                @Nullable
                private Duration queryTimeout;
                @Nullable
                private Integer fetchSize;
                @Nullable
                private Integer maxRows;
                @Nullable
                private Integer parsedSqlCacheCapacity;

                private Builder(@NonNull DataSource dataSource) {
                        this.dataSource = requireNonNull(dataSource);
                        this.databaseType = null;
                        this.metricsCollector = null;
                        this.queryTimeout = null;
                        this.fetchSize = null;
                        this.maxRows = null;
                        this.parsedSqlCacheCapacity = null;
                }

                /**
                 * Overrides automatic database type detection.
                 * <p>
                 * If {@code null}, the database type is detected lazily when database-type-specific behavior is first needed.
                 * Supplying a non-null value avoids automatic detection and its metadata lookup entirely.
                 *
                 * @param databaseType the database type to use (null to enable auto-detection)
                 * @return this {@code Builder}, for chaining
                 * @since 4.0.0
                 */
                @NonNull
                public Builder databaseType(@Nullable DatabaseType databaseType) {
                        this.databaseType = databaseType;
                        return this;
                }

                /**
                 * Configures the database time zone Pyranid should use when converting zone-less temporal values.
                 * <p>
                 * This value is used when mapping {@code TIMESTAMP} values to instant-based Java types, and when binding
                 * {@link java.time.Instant} or {@link java.time.OffsetDateTime} parameters to known {@code TIMESTAMP}
                 * targets. It also applies to ambiguous timestamp bindings if
                 * {@link #ambiguousTimestampBindingStrategy(AmbiguousTimestampBindingStrategy)} is configured with
                 * {@link AmbiguousTimestampBindingStrategy#TIMESTAMP_WITHOUT_TIME_ZONE}.
                 * <p>
                 * If {@code null}, Pyranid uses {@link ZoneId#systemDefault()}.
                 *
                 * @param timeZone database time zone to use, or {@code null} for the JVM default zone
                 * @return this {@code Builder}, for chaining
                 * @since 3.0.0
                 */
                @NonNull
                public Builder timeZone(@Nullable ZoneId timeZone) {
                        this.timeZone = timeZone;
                        return this;
                }

                /**
                 * Configures how Pyranid binds {@link java.time.Instant} and {@link java.time.OffsetDateTime} parameters
                 * when JDBC parameter metadata cannot identify whether the target is {@code TIMESTAMP} or
                 * {@code TIMESTAMP WITH TIME ZONE}.
                 * <p>
                 * The default, {@link AmbiguousTimestampBindingStrategy#TIMESTAMP_WITH_TIME_ZONE}, is appropriate for
                 * timestamp-with-time-zone targets. Use
                 * {@link AmbiguousTimestampBindingStrategy#TIMESTAMP_WITHOUT_TIME_ZONE} for drivers or proxies that
                 * cannot provide identifying parameter metadata when your target columns are zone-less {@code TIMESTAMP}
                 * values that should be interpreted in {@link #timeZone(ZoneId)}.
                 *
                 * @param ambiguousTimestampBindingStrategy strategy to use, or {@code null} for the default
                 * @return this {@code Builder}, for chaining
                 * @since 4.2.0
                 */
                @NonNull
                public Builder ambiguousTimestampBindingStrategy(@Nullable AmbiguousTimestampBindingStrategy ambiguousTimestampBindingStrategy) {
                        this.ambiguousTimestampBindingStrategy = ambiguousTimestampBindingStrategy;
                        return this;
                }

                @NonNull
                public Builder instanceProvider(@Nullable InstanceProvider instanceProvider) {
                        this.instanceProvider = instanceProvider;
                        return this;
                }

                @NonNull
                public Builder preparedStatementBinder(@Nullable PreparedStatementBinder preparedStatementBinder) {
                        this.preparedStatementBinder = preparedStatementBinder;
                        return this;
                }

                @NonNull
                public Builder resultSetMapper(@Nullable ResultSetMapper resultSetMapper) {
                        this.resultSetMapper = resultSetMapper;
                        return this;
                }

                /**
                 * Configures the statement logger for the {@link Database} being built.
                 * <p>
                 * {@link StatementLogger} failures are fail-fast: a logger exception can make a successful statement operation throw,
                 * and inside a Pyranid transaction it participates in normal rollback handling. If the database statement itself failed,
                 * logger failures are suppressed onto the primary statement failure.
                 *
                 * @param statementLogger statement logger to use, or {@code null} for a no-op logger
                 * @return this {@code Builder}, for chaining
                 */
                @NonNull
                public Builder statementLogger(@Nullable StatementLogger statementLogger) {
                        this.statementLogger = statementLogger;
                        return this;
                }

                /**
                 * Configures the redactor used for non-secure parameters in diagnostics.
                 * <p>
                 * {@link SecureParameter} values always render via {@link SecureParameter#getMask()} and are never passed to this
                 * redactor. Batch executions render a bounded batch summary instead of invoking the redactor for each batch value.
                 * Specify {@code null} or omit this setter to render non-secure, non-batch values verbatim.
                 *
                 * @param parameterRedactor parameter redactor to use, or {@code null} for the default
                 * @return this {@code Builder}, for chaining
                 * @since 4.4.0
                 */
                @NonNull
                public Builder parameterRedactor(@Nullable ParameterRedactor parameterRedactor) {
                        this.parameterRedactor = parameterRedactor;
                        return this;
                }

                /**
                 * Configures the metrics collector for the {@link Database} being built.
                 * <p>
                 * Like all {@code Database} configuration, this value is fixed at {@link #build()} time. Specify {@code null}
                 * or omit this setter to disable metrics collection.
                 *
                 * @param metricsCollector metrics collector to use, or {@code null} to disable
                 * @return this {@code Builder}, for chaining
                 * @since 4.2.0
                 */
                @NonNull
                public Builder metricsCollector(@Nullable MetricsCollector metricsCollector) {
                        this.metricsCollector = metricsCollector;
                        return this;
                }

                /**
                 * Configures a database-wide JDBC query timeout default.
                 * <p>
                 * This maps to {@link java.sql.Statement#setQueryTimeout(int)}. {@code null} leaves the timeout unset.
                 * {@link Duration#ZERO} disables the JDBC timeout. Positive sub-second durations are rounded up to one second
                 * because JDBC accepts whole seconds. Per-query {@link Query#queryTimeout(Duration)} settings override this value,
                 * and {@link Query#customize(PreparedStatementCustomizer)} can override both settings before parameter binding.
                 *
                 * @param queryTimeout timeout to apply by default, or {@code null} to leave unset
                 * @return this {@code Builder}, for chaining
                 * @since 4.2.0
                 */
                @NonNull
                public Builder queryTimeout(@Nullable Duration queryTimeout) {
                        this.queryTimeout = validateQueryTimeout(queryTimeout);
                        return this;
                }

                /**
                 * Configures a database-wide JDBC fetch size default.
                 * <p>
                 * This maps to {@link java.sql.Statement#setFetchSize(int)}. {@code null} leaves the fetch size unset. A value
                 * of {@code 0} uses the driver's default fetch-size behavior. Per-query {@link Query#fetchSize(Integer)}
                 * settings override this value, and {@link Query#customize(PreparedStatementCustomizer)} can override both
                 * settings before parameter binding.
                 *
                 * @param fetchSize fetch size to apply by default, or {@code null} to leave unset
                 * @return this {@code Builder}, for chaining
                 * @since 4.2.0
                 */
                @NonNull
                public Builder fetchSize(@Nullable Integer fetchSize) {
                        this.fetchSize = validateNonNegativeStatementSetting("fetchSize", fetchSize);
                        return this;
                }

                /**
                 * Configures a database-wide JDBC maximum row count default.
                 * <p>
                 * This maps to {@link java.sql.Statement#setMaxRows(int)}. {@code null} leaves the maximum row count unset. A
                 * value of {@code 0} disables the JDBC row limit. Per-query {@link Query#maxRows(Integer)} settings override
                 * this value, and {@link Query#customize(PreparedStatementCustomizer)} can override both settings before parameter
                 * binding.
                 *
                 * @param maxRows maximum rows to apply by default, or {@code null} to leave unset
                 * @return this {@code Builder}, for chaining
                 * @since 4.2.0
                 */
                @NonNull
                public Builder maxRows(@Nullable Integer maxRows) {
                        this.maxRows = validateNonNegativeStatementSetting("maxRows", maxRows);
                        return this;
                }

                /**
                 * Configures the size of the parsed SQL cache.
                 * <p>
                 * A value of {@code 0} disables caching. A value of {@code null} uses the default size.
                 *
                 * @param parsedSqlCacheCapacity cache size (0 disables caching, null uses default)
                 * @return this {@code Builder}, for chaining
                 */
                @NonNull
                public Builder parsedSqlCacheCapacity(@Nullable Integer parsedSqlCacheCapacity) {
                        if (parsedSqlCacheCapacity != null && parsedSqlCacheCapacity < 0)
                                throw new IllegalArgumentException("parsedSqlCacheCapacity must be >= 0");

                        this.parsedSqlCacheCapacity = parsedSqlCacheCapacity;
                        return this;
                }

                @NonNull
                public Database build() {
                        return new Database(this);
                }
        }

        @ThreadSafe
        static class DatabaseOperationResult {
                @Nullable
                private final Duration executionDuration;
                @Nullable
                private final Duration resultSetMappingDuration;
                @NonNull
                private final StatementResult statementResult;

                public DatabaseOperationResult(@Nullable Duration executionDuration,
                                                                                                                                         @Nullable Duration resultSetMappingDuration) {
                        this(executionDuration, resultSetMappingDuration, StatementResult.empty());
                }

                public DatabaseOperationResult(@Nullable Duration executionDuration,
                                                                                                                                         @Nullable Duration resultSetMappingDuration,
                                                                                                                                         @NonNull StatementResult statementResult) {
                        this.executionDuration = executionDuration;
                        this.resultSetMappingDuration = resultSetMappingDuration;
                        this.statementResult = requireNonNull(statementResult);
                }

                @NonNull
                public Optional<Duration> getExecutionDuration() {
                        return Optional.ofNullable(this.executionDuration);
                }

                @NonNull
                public Optional<Duration> getResultSetMappingDuration() {
                        return Optional.ofNullable(this.resultSetMappingDuration);
                }

                @NonNull
                public StatementResult getStatementResult() {
                        return this.statementResult;
                }
        }

        @NotThreadSafe
        static class ResultHolder<T> {
                T value;
        }

        enum DatabaseOperationSupportStatus {
                UNKNOWN,
                YES,
                NO
        }

}