[#888] Name the trees of a JDBC backend from a catalog in the database (#893)

Valery Kharseko

9 hours ago 2b8612f7fa2f5e00dff4ff2d8ac912c833e1f7ab

[#888] Name the trees of a JDBC backend from a catalog in the database (#893)

 opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java

@@ -37,6 +37,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.LongFunction;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@@ -235,6 +236,54 @@
    private static volatile Executor closer = DIRECT_EXECUTOR;

    /**
     * Returns the bound of one attempt to establish a connection, as configured by the {@value
     * #CONNECT_TIMEOUT_PROPERTY} system property; 0 for the operator asking for no bound of its own.
     * A value beyond what a millisecond bound can carry is taken down to it: three of the four
     * dialects state their properties in milliseconds, and a value that saturates the conversion
     * bounds nothing.
     * <p>
     * Read here rather than at each connect so that every connection this backend establishes is
     * bounded by the same configured value - the borrows of this pool and the connection {@code
     * JDBCStorage} opens outside it for the tree catalog of a backend (#888) alike. A connect
     * bounded tighter than the login of the deployment takes is a backend that stops opening, and
     * one place to read the property is what keeps the two from drifting apart.
     */
    static long getConnectTimeoutSeconds() {
        return Math.min(getNonNegativeProperty(CONNECT_TIMEOUT_PROPERTY, DEFAULT_CONNECT_TIMEOUT_SECONDS, "s"),
            Integer.MAX_VALUE / 1000);
    }

    /**
     * Returns the deadline of a whole borrow, as configured by the {@value #POOL_TIMEOUT_PROPERTY}
     * system property; 0 for the operator asking for no deadline at all.
     * <p>
     * Read here for the reason the bound of a connect is: it is what a database taking no
     * connection for the moment is waited out for, and the connection {@code JDBCStorage} opens
     * outside this pool for the tree catalog of a backend (#888) waits it out for exactly as long -
     * one property, one meaning, whichever of the two is asking.
     */
    static long getPoolTimeoutSeconds() {
        return getNonNegativeProperty(POOL_TIMEOUT_PROPERTY, DEFAULT_POOL_TIMEOUT_SECONDS, "s");
    }

    /**
     * The moment a borrow of this length gives up, or {@link Long#MAX_VALUE} where it gives up
     * never - a property of 0, and a value so large that the milliseconds of it would overflow.
     * <p>
     * The sum is guarded and not only the product: a value under the clamp above but large enough
     * that the moment it names is past the end of the epoch would wrap to a deadline already behind
     * us, and a borrow configured to wait practically forever would give up on its first retryable
     * failure - the opposite of what was asked for.
     */
    static long deadlineOf(long startedAt, long poolTimeoutSeconds) {
        if (poolTimeoutSeconds == 0 || poolTimeoutSeconds >= Long.MAX_VALUE / 1000) {
            return Long.MAX_VALUE;
        }
        final long deadline = startedAt + poolTimeoutSeconds * 1000;
        return deadline < startedAt ? Long.MAX_VALUE : deadline;
    }

    /**
     * Returns the time after which an idle pooled connection is closed, as configured by the
     * {@value #TTL_PROPERTY} system property. An invalid value is ignored in favor of the default.
     * <p>
@@ -399,7 +448,10 @@
        if (borrowers <= pool.max()) {
            return;
        }
        final long poolTimeoutSeconds = getNonNegativeProperty(POOL_TIMEOUT_PROPERTY, DEFAULT_POOL_TIMEOUT_SECONDS, "s");
        // through the helper the borrows and the catalog connect both read it by: the same property
        // has to mean the same thing wherever it is asked, and a clamp that helper grows the day the
        // deadline needs one - getConnectTimeoutSeconds() already has one - must not be missed here
        final long poolTimeoutSeconds = getPoolTimeoutSeconds();
        final String wait = poolTimeoutSeconds == 0
            ? "waits for one to be returned for as long as that takes"
            : "waits up to " + poolTimeoutSeconds + "s for one to be returned and fails if none is";
@@ -1079,14 +1131,11 @@
        final Pool pool = poolOf(connectionString);
        final ConnectDialect dialect = ConnectDialect.of(connectionString);
        reportUnknownDialect(connectionString, dialect);
        final long connectTimeoutSeconds = Math.min(
            getNonNegativeProperty(CONNECT_TIMEOUT_PROPERTY, DEFAULT_CONNECT_TIMEOUT_SECONDS, "s"),
            Integer.MAX_VALUE / 1000);
        final long poolTimeoutSeconds = getNonNegativeProperty(POOL_TIMEOUT_PROPERTY, DEFAULT_POOL_TIMEOUT_SECONDS, "s");
        final long connectTimeoutSeconds = getConnectTimeoutSeconds();
        final long poolTimeoutSeconds = getPoolTimeoutSeconds();
        final long ttlMillis = getCacheTtlMillis();
        final long startedAt = System.currentTimeMillis();
        final long deadline = (poolTimeoutSeconds == 0 || poolTimeoutSeconds >= Long.MAX_VALUE / 1000)
            ? Long.MAX_VALUE : startedAt + poolTimeoutSeconds * 1000;
        final long deadline = deadlineOf(startedAt, poolTimeoutSeconds);
        // A thread already holding a connection is not made to wait for one: the two are held at
        // the same time, so waiting for the first to come back would wait for itself.
        final boolean reentrant = pool.heldByCurrentThread();
@@ -1493,16 +1542,71 @@
    // is indistinguishable from a hang. Throttled, since every operation of the backend borrows
    // through here and would otherwise log a copy of its own.
    private static void warnStall(String connectionString, int attempts, long startedAt, SQLException cause) {
        warnStall(connectionString, "", startedAt,
            waitedMs -> stallMessage(connectionString, attempts, waitedMs, cause));
    }

    /**
     * The same for a connect this class makes for somebody outside the pool - the connection the
     * tree catalog of a backend is written on (#888) - which waits for no pooled connection and
     * must not be described as one.
     * <p>
     * Throttled apart from the borrows of the same url as well as worded apart from them: the two
     * stall on the same database for the same reason, so a borrow that warned a moment ago would
     * otherwise silence the connect that is about to fail - the one of the two an operator has no
     * other line about.
     */
    static void warnStallOutsidePool(String connectionString, String what, int attempts, long startedAt,
            SQLException cause) {
        warnStall(connectionString, "|" + what, startedAt,
            waitedMs -> outsidePoolStallMessage(connectionString, what, attempts, waitedMs, cause));
    }

    private static void warnStall(String connectionString, String throttleKeySuffix, long startedAt,
            LongFunction<String> message) {
        final long now = System.currentTimeMillis();
        if (stallWarningDue(connectionString, throttleKeySuffix, startedAt, now)) {
            logger.warn(LocalizableMessage.raw("%s", message.apply(now - startedAt)));
        }
    }

    /**
     * Whether a stall of this wait is to be reported now: a wait shorter than
     * {@link #STALL_WARNING_AFTER_MS} is no stall yet, and one already reported for this url within
     * {@link #STALL_WARNING_INTERVAL_MS} is not reported again - every worker thread of a server
     * meets a database taking no connection at the same moment, and one line an interval is what
     * an operator can read.
     * <p>
     * Built apart from the logging of it for the reason {@link #stallMessage} is: what it has to
     * keep is a rule a test can hold it to, and the shipped path reaches the throttle only where a
     * connect really has stalled for a second. The suffix is what keeps the two waits apart - a
     * borrow of the pool and the connect the tree catalog of a backend is made on (#888) stall on
     * the same database for the same reason, and a borrow that reported a moment ago must not
     * silence the connect that is about to fail, which has no other line about it at all.
     * <p>
     * Filing the moment is part of deciding it, so that two threads asking at once report once.
     */
    static boolean stallWarningDue(String connectionString, String throttleKeySuffix, long startedAt, long now) {
        if (now - startedAt < STALL_WARNING_AFTER_MS) {
            return;
            return false;
        }
        final AtomicLong lastOfThisUrl =
            lastStallWarning.computeIfAbsent(safeUrl(connectionString), url -> new AtomicLong());
            lastStallWarning.computeIfAbsent(safeUrl(connectionString) + throttleKeySuffix, url -> new AtomicLong());
        final long last = lastOfThisUrl.get();
        if (now - last >= STALL_WARNING_INTERVAL_MS && lastOfThisUrl.compareAndSet(last, now)) {
            logger.warn(LocalizableMessage.raw("%s", stallMessage(connectionString, attempts, now - startedAt, cause)));
        }
        return now - last >= STALL_WARNING_INTERVAL_MS && lastOfThisUrl.compareAndSet(last, now);
    }

    /**
     * The stall of a connect made outside the pool, as it reaches the log. Built apart from the
     * logging of it for the reason {@link #stallMessage} is: the rule it has to keep - neither the
     * connection string nor the message of the driver reaches a log as it stands - is a rule a test
     * can hold it to.
     */
    static String outsidePoolStallMessage(String connectionString, String what, int attempts, long waitedMs,
            SQLException cause) {
        return String.format("%s takes no further connection: the %s connection of this backend is opened outside the"
            + " pool and has been retrying for %d ms (%d attempts), last error: %s", safeUrl(connectionString), what,
            waitedMs, attempts, redact(cause.getMessage(), connectionString));
    }

    /**

 opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java

@@ -25,6 +25,7 @@
import org.forgerock.opendj.config.server.ConfigurationChangeListener;
import org.forgerock.opendj.ldap.ByteSequence;
import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.backends.pluggable.spi.*;
import org.opends.server.core.ServerContext;
@@ -36,6 +37,7 @@

import java.io.Closeable;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.sql.*;
@@ -823,31 +825,37 @@
        // that it is not reissued for every tree on every open; disabling and re-enabling the
        // backend is the way to try again once the privilege has been granted
        unstampableTrees.clear();
        // what this storage knows of its catalog holds no longer than the open it learnt it in: the
        // table may well be gone by the next one, dropped by an offline tool run in the meantime
        catalogTableOpened=false;
        enrolledTrees.clear();
        // A closed backend has no use for its connections. They used to stay open - close() only
        // flipped the status - so disabling or removing a JDBC backend left them behind, and with
        // nothing left to expire the pool entry they could stay open for good (issue #878).
        releasePool();
    }

    // The trees this storage has taken an interest in, and the tables they map to. listTrees() -
    // and through it removeStorageFiles() - reads this, so a tree only belongs here once this
    // backend uses it: see toTableName() below for the trees that are merely asked about.
    // The trees this storage has taken an interest in, and the tables they map to: a memo, so that
    // naming the table of a tree costs a map lookup rather than a digest. What a backend owns is
    // recorded in its catalog and not here (#888) - listTrees() and removeStorageFiles() read that
    // - but the distinction the two names below draw is kept all the same: a tree merely asked
    // about is not one this storage has taken an interest in, and it stays out of the memo.
    final LoadingCache<TreeName,String> tree2table = Caffeine.newBuilder()
        .build(JDBCStorage::toTableName);

    /**
     * The table a tree name maps to. A pure function of the name, so that a tree can be read
     * without being entered into tree2table: the compressed schema reads the tree its definitions
     * used to be shared under (#873), a tree this backend does not own, and removeStorageFiles()
     * drops every table tree2table names.
     * The table a tree name maps to. A pure function of the name, so that a tree can be read without
     * being entered into tree2table: the compressed schema reads the tree its definitions used to be
     * shared under (#873), which is a tree this backend does not own.
     * <p>
     * Which of the two a statement takes therefore says who owns the tree it names: a path that
     * creates or writes one - openTree(), clearTree(), deleteTree(), put(), update(), delete() -
     * takes the enrolling {@link #getTableName(TreeName)}, and a read-only path - read(),
     * getRecordCount(), isExistsTable() and the cursor - takes {@link #readTableName(TreeName)},
     * which computes this only for a tree that is not enrolled already. Every tree this backend
     * owns passes through openTree(name, true) as it is opened, so listTrees() still names the
     * complete owned set.
     * Which of the two names a statement takes therefore says whether this backend is claiming the
     * tree it names: a path that creates or writes one - openTree(), clearTree(), deleteTree(), put(),
     * update(), delete() - takes the enrolling {@link #getTableName(TreeName)}, and a path that only
     * asks - read(), getRecordCount(), isExistsTable(), the cursor, and the read of what the catalog
     * records - takes {@link #readTableName(TreeName)}, which computes this only for a tree that is not
     * enrolled already. What a clear may drop is decided by the catalog of the backend (#888) and no
     * longer by this memo, so an entry of it puts no table up for removal; the two names are what keeps
     * the memo an account of the trees this backend claims all the same.
     */
    static String toTableName(TreeName treeName) {
        try {
@@ -870,6 +878,85 @@
    }

    /**
     * The pseudo base DN of the tree naming the trees of a backend. Every real tree of a backend is
     * named after an entry container, whose prefix is a normalized DN and so always holds a "=",
     * which an identifier of this form cannot collide with.
     */
    static final String CATALOG_BASE_DN="opendj_catalog";

    /**
     * The base DN the compressed schema trees were named under before #881 gave each backend a pair
     * of its own. It carries no backend qualifier, so on a database addressed by several backends -
     * which nothing forbids (#873) - that pair of trees is the same pair for all of them, and a
     * backend must not put a tree another one may be the owner of up for removal. It is the pair
     * {@code PersistentCompressedSchema} migrates from and never writes to again, and it is left
     * exactly where it lies: the definitions of a backend that has not been started since the
     * upgrade are still in it. The pair each backend owns is named after its backend id, is under no
     * such literal, and is enrolled like any other tree.
     */
    static final String SHARED_COMPRESSED_SCHEMA_BASE_DN="compressed_schema";

    /**
     * The pair named under {@link #SHARED_COMPRESSED_SCHEMA_BASE_DN}, spelled out here because the
     * names are private to {@code PersistentCompressedSchema} - where they are the LEGACY_ pair of
     * #881. They are never enrolled, so nothing but this constant can name them - and a tool asking
     * a backend what trees it holds has to be told about them all the same, which is what {@link
     * #listTrees()} uses this for.
     */
    static final List<TreeName> SHARED_COMPRESSED_SCHEMA_TREES=Collections.unmodifiableList(Arrays.asList(
        new TreeName(SHARED_COMPRESSED_SCHEMA_BASE_DN, "compressed_attributes"),
        new TreeName(SHARED_COMPRESSED_SCHEMA_BASE_DN, "compressed_object_classes")));

    /**
     * The tree naming the trees this backend owns: one row per tree, the tree name as its key and the
     * table holding that tree as its value.
     * <p>
     * A table is named after the hash of its tree name, so the catalog of a database can neither be
     * filtered by a per-backend prefix nor read back into a {@link TreeName}. Without a record of its
     * own a backend can therefore only name the trees this very process has already touched - which
     * is precisely what {@link #removeStorageFiles()} cannot have, running as it does before the root
     * container is open. In the offline {@code import-ldif} nothing has touched a tree at all, so
     * {@code --clearBackend} used to clear nothing whatsoever (#888).
     * <p>
     * The catalog is per backend and named after the backend id alone: a process that has opened
     * nothing can still find its table, and backends sharing one database URL - which nothing
     * forbids (#873) - never name each other's trees. The id goes in escaped, for the reason {@link
     * #escapedBackendId} states: a name that does not survive being read back is a table of this
     * backend that its own clear cannot recognize.
     */
    TreeName getCatalogTree() {
        return new TreeName(CATALOG_BASE_DN, escapedBackendId());
    }

    /**
     * Whether the table of the catalog was created, or found, by this storage. A tree is enrolled on
     * every open - about 25 of them for a stock suffix - and asking the catalog whether the table is
     * there would cost a metadata round trip per tree.
     */
    private volatile boolean catalogTableOpened=false;

    /**
     * Serializes the one step above: two transactions opening trees at the same time would otherwise
     * both find the table of the catalog absent and both create it, the second failing the open it
     * belongs to. Held across the lookup and the statement that answer it, and across nothing else.
     */
    private final Object catalogLock=new Object();

    /**
     * The trees the catalog already records at the table this version would record them at, read
     * from it when this storage first opens it and added to as it enrols. A tree named here needs no
     * row written for it: the row would be the one that is already there, and writing one is a
     * statement and a commit on a connection this backend then has to have opened - a stock suffix
     * has about 25 trees, and every open after the first enrols none of them.
     * <p>
     * A row recording another table than {@link #getTableName} would give is not in here: what a
     * removal drops is the table the row records, so a row of a version naming its tables otherwise
     * has to be rewritten rather than trusted. Held no longer than the open it was read in, like
     * {@link #catalogTableOpened}, and given up whenever the catalog itself is.
     */
    private final Set<TreeName> enrolledTrees=ConcurrentHashMap.newKeySet();

    /**
     * The table a tree name maps to, for a statement that only reads it. Answered from the memo of
     * {@link #getTableName(TreeName)} where the tree is in it, and computed without being put there
     * otherwise.
@@ -894,10 +981,10 @@
     */
    static String storedIdentifier(DatabaseMetaData metaData, String name) throws SQLException {
        if (metaData.storesUpperCaseIdentifiers()) {
            return name.toUpperCase();
            return name.toUpperCase(Locale.ROOT);
        }
        if (metaData.storesLowerCaseIdentifiers()) {
            return name.toLowerCase();
            return name.toLowerCase(Locale.ROOT);
        }
        return name;
    }
@@ -1049,7 +1136,7 @@
    boolean isMysqlBackslashEscape(Connection con) throws SQLException {
        try (final PreparedStatement statement=con.prepareStatement("select @@sql_mode")) {
            final String sqlMode=executeResultSet(statement, rs -> rs.next() ? rs.getString(1) : null);
            return sqlMode==null || !sqlMode.toUpperCase().contains("NO_BACKSLASH_ESCAPES");
            return sqlMode==null || !sqlMode.toUpperCase(Locale.ROOT).contains("NO_BACKSLASH_ESCAPES");
        }
    }

@@ -1084,6 +1171,417 @@
        return con;
    }

    /**
     * A connection of its own for the catalog of a backend, outside the pool for the reason a stamp
     * connection is: the caller of openTree() is inside a transaction and holding a pooled connection
     * already, and a pool that cannot open a second one waits for a peer to return one - which here is
     * the very thread that is waiting.
     * <p>
     * It is established the way a pooled connection is and not the way a stamp connection is: the
     * bounds of {@link CachedConnection.ConnectDialect} rather than of {@link Dialect}, so that a
     * login which never answers is bounded, a bound the administrator set in the connection string is
     * left exactly as they set it, and the read bound of the login is lifted as soon as the login is
     * through (#872). A stamp is a diagnostic aid and gives up rather than queue behind another
     * session; a catalog row is the state a clear reads, and it waits for its lock rather than dying
     * on a read bound. The isolation is the pool's for the same reason: this connection issues the
     * ordinary DML of this class, and the repeatable read a mysql server defaults to gap-locks a
     * catalog two transactions enrol into.
     * <p>
     * The bound of the connect is the one the pool bounds its own connects by, read from {@link
     * CachedConnection#CONNECT_TIMEOUT_PROPERTY} where an operator set it: a login of this database
     * takes what it takes whoever is asking, so a deployment which had to raise that property must not
     * meet a bound of this code's own here - a connect failing where the pooled one beside it succeeds
     * is a backend that stops opening on an installation that opened before this connection existed. A
     * property of 0 is the operator asking for no bound of the connect, and it is honoured here as it
     * is by the pool. What does bound an attempt besides is the deadline of the retry below, which is
     * the pool's own rule and applies to a borrow in exactly the same way; it is no bound of this
     * code's own choosing.
     * <p>
     * The deadline of the whole thing is the pool's as well, {@link
     * CachedConnection#POOL_TIMEOUT_PROPERTY}: a database that takes no connection <em>for the
     * moment</em> - at its connection limit with one of ours on its way back to the pool, or still
     * recovering - is waited out here exactly as a borrow waits it out, by the predicate the pool
     * decides that by ({@link CachedConnection#isWorthRetrying}) and with the same backoff. Without
     * it this connect makes one attempt where the borrow beside it makes many, and loses a race the
     * pooled connection of the very same operation wins. Everything else - a password that is not
     * accepted, a database that is down, a driver that is not on the classpath - is reported to the
     * caller rather than retried behind its back.
     * <p>
     * What this deadline is not is the deque of the pool: the caller of {@code openTree()} is holding
     * a pooled connection already, so waiting for a peer to return one would be waiting for the very
     * thread that is waiting. It is the pool's retry that is wanted here and not its queue, which is
     * why the loop below is its own rather than a borrow of {@link CachedConnection#getConnection}.
     * What one attempt is, is the login and the set-up behind it, exactly as an attempt of a borrow is
     * ({@code CachedConnection.connect}): a session the server takes and then kills off answers the
     * first statement of the set-up rather than the login, and it is the same refusal either way.
     * <p>
     * That is also what this wait is weaker than a borrow at, and it is worth writing down rather than
     * leaving to be discovered: a borrow can be answered by a peer handing a connection back, while
     * nothing here can be answered by anything but a new login. Against a server at its connection
     * limit whose remaining slots this backend's own pool is holding idle, the borrows of that pool
     * clear and this does not - it waits out the deadline and reports the refusal. The deadline is
     * therefore what bounds it, and a deployment which has set both properties to 0 has asked for a
     * wait with no end to it here as much as in the pool.
     * <p>
     * One attempt is bounded by the configured connect timeout and by what is left of that deadline,
     * whichever is the shorter, exactly as an attempt of a borrow is: an attempt left to run its own
     * bound out past the deadline would overrun it by a whole connect timeout, and turning the
     * per-attempt bound off must not turn the deadline off with it. So the connect property of 0 that
     * the round before this one made honoured is the operator asking for no bound <em>of their own</em>
     * here as it is in the pool, and what is left unbounded by both properties at 0 is left unbounded
     * here too - a login the database accepts and never finishes then parks the transaction that asked
     * for it. A borrow of the pool parks in exactly the same way, on exactly the same pair of settings.
     * It does not park the rest of this storage: {@code openCatalog()} establishes this connection
     * before it takes its lock, for that very reason.
     * <p>
     * What every wait here does hold is the caller: this runs inside the write transaction that reached
     * {@code openTree}, so a retry that waits out a database refusing connections holds that
     * transaction's pooled connection, the permit of the pool that connection carries (#878) and every
     * lock the transaction has already taken, for as long as it waits. On a stock suffix {@code
     * RootContainer.open()} is one such write over every tree of the backend.
     * <p>
     * And what it spends besides is the window {@link #write} bounds its own replay by, which is the
     * shorter of the two by default - ten seconds against a minute - and is <em>spent</em> by this wait
     * rather than added to it: this loop runs inside one attempt of that one. A refusal {@code write()}
     * would replay - mysql answers its connection limit with {@code 08004}, postgres reports a database
     * still coming up as {@code 57P03}, and both are read as a connection this backend lost - waited out
     * here for a minute reaches that loop with its window six times over, so it is thrown unreplayed:
     * the retry would have cost the caller the very replay it had before there was any retry here at
     * all. So the deadline is the shorter of {@link CachedConnection#POOL_TIMEOUT_PROPERTY} and what is
     * left of that window, taken from the caller by {@link CatalogSession#boundedAlsoBy}. The window is
     * not something the property could express: lowering it under ten seconds shortens every borrow of
     * the pool with it. A path carrying no such window - the importer, which has no replay above it -
     * waits the property out in full, and a deployment which cannot afford a minute of that sets the
     * property to what it can afford, the same property bounding the same wait as it bounds a borrow.
     * <p>
     * What the window does <em>not</em> bound is one attempt, which is taken from the deadline of the
     * pool as it always was. The two are different questions: the window says how long it is worth
     * waiting before handing the failure to a loop that can still replay it, while a login takes what
     * this database takes whoever is asking. Cut to what is left of a ten second window, a deployment
     * that raised {@link CachedConnection#CONNECT_TIMEOUT_PROPERTY} to two minutes because its login
     * needs them would meet a catalog connect failing where the pooled connection beside it succeeds -
     * the backend that stops opening. So one slow attempt may outlast the window, exactly as one slow
     * conflict outlasts it in {@link #write} itself; what may not is a second attempt begun after the
     * window has already run out, which is a wait bought with a replay that no longer exists.
     *
     * @param budgetDeadline the moment the replay window of the caller runs out, as {@link
     *        System#currentTimeMillis()} reads it, or {@link Long#MAX_VALUE} where nothing above this
     *        connect replays - the same "no deadline at all" this class reads out of {@link
     *        CachedConnection#deadlineOf}, so that the shorter of the two is a plain {@code min}.
     */
    Connection newCatalogConnection(long budgetDeadline) throws SQLException {
        // poolKey() rather than the configuration as it stands, for the reason newStampConnection()
        // gives: this connection is not pooled, but it is a connection to the database of this
        // storage, and db-directory may be changed on a running backend. Reading it again here would
        // write the catalog of this backend into whichever database the configuration names now,
        // while its tables are created, read and dropped over the connection open() registered -
        // rows in one database and tables in another, which is #888 again by another route (#878)
        final String connectionString=poolKey();
        final CachedConnection.ConnectDialect dialect=CachedConnection.ConnectDialect.of(connectionString);
        final long connectTimeoutSeconds=CachedConnection.getConnectTimeoutSeconds();
        final long poolTimeoutSeconds=CachedConnection.getPoolTimeoutSeconds();
        final long startedAt=System.currentTimeMillis();
        final long poolDeadline=CachedConnection.deadlineOf(startedAt, poolTimeoutSeconds);
        // the deadline of the whole wait, which is the shorter of the pool's own and what is left of
        // the replay window of the caller. The bound of one attempt below is taken from the pool's
        // alone, deliberately: a login the operator bounded at two minutes because that is what this
        // database takes must not be cut to what is left of a ten second window - that is the connect
        // dying where the pooled one beside it succeeds, which is a backend that stops opening. One
        // slow attempt may still outlast the window, exactly as one slow conflict does; what may not
        // is a second attempt begun after the window has run out, which is a wait for nothing
        final long deadline=Math.min(poolDeadline, budgetDeadline);
        long backoffMs=0;
        int attempts=0;
        while (true) {
            attempts++;
            try {
                // the bound of one attempt and not of the whole wait, the way the pool bounds its own:
                // an attempt left to run its bound out past the deadline would overrun it by a full
                // connect timeout, and turning the per-attempt bound off must not turn this one off.
                // Taken from the deadline of the pool and not from the shorter of the two: see above
                return connectCatalog(connectionString, dialect,
                    CachedConnection.attemptSeconds(connectTimeoutSeconds, poolDeadline));
            }catch (SQLException e) {
                if (!CachedConnection.isWorthRetrying(e, dialect)) {
                    // redacted the way the pool redacts the failure of its own connects: a driver renders
                    // the connection string it could not use into its message as readily as not, and the
                    // connection string of this backend carries the password of the account it works as.
                    // This failure is reported in full - ERR_OPEN_ENV_FAIL, or the log of a clear
                    throw CachedConnection.reported(e, connectionString);
                }
                final long now=System.currentTimeMillis();
                final long remaining=deadline-now;
                if (remaining<=0) {
                    // which of the two bounds ended it, so that an operator reading the line knows whether
                    // the property is the thing to raise: where the replay window of the caller is the
                    // shorter one, raising the property moves nothing
                    throw catalogConnectTimedOut(connectionString, poolTimeoutSeconds,
                        deadline==budgetDeadline, now-startedAt, attempts, e);
                }
                CachedConnection.warnStallOutsidePool(connectionString, "tree catalog", attempts, startedAt, e);
                backoffMs=Math.min(backoffMs==0 ? 1 : backoffMs*2, CachedConnection.MAX_BACKOFF_MS);
                try {
                    Thread.sleep(Math.min(backoffMs, remaining));
                }catch (InterruptedException interrupted) {
                    // the flag is put back - Thread.sleep() clears it, and every frame above this one reads
                    // it to decide whether to unwind - and the wait is over: whoever asked this thread to
                    // stop is not answered by going on to sleep out the rest of a pool timeout. The driver
                    // failure is what this reports, it being the reason there was anything to wait for, and
                    // the interrupt is carried on it as suppressed so that a connect cut short by a
                    // shutdown is not read off the log as a database that would not take a connection
                    Thread.currentThread().interrupt();
                    final SQLException reported=CachedConnection.reported(e, connectionString);
                    reported.addSuppressed(interrupted);
                    throw reported;
                }
            }catch (RuntimeException e) {
                // a driver reporting a connect it will not make as an unchecked failure names the
                // connection string just as readily, and it is not one of the two states a retry waits
                // out: reported and handed on, exactly as the pool hands its own on. reportedUnchecked()
                // answers with the original where it holds no credential, so nothing of a plain
                // programming error is hidden by this
                final Exception reported=CachedConnection.reportedUnchecked(e, connectionString);
                if (reported instanceof SQLException) { // redacted, and reported as the connect failure it is
                    throw (SQLException) reported;
                }
                throw (RuntimeException) reported; // the original: it holds no credential of this backend
            }
        }
    }

    /**
     * The failure of a catalog connect that was worth retrying and ran the deadline out: a timeout by
     * type, so that a caller can tell it from the first refusal, and carrying the state and the vendor
     * code of the last failure of the driver rather than one of its own.
     * <p>
     * Not the {@code 08001} the pool answers a borrow of this shape with, and the difference is not
     * cosmetic: this failure is raised inside {@link #write}, whose classification reads every state
     * of class {@code 08} as a connection the database dropped ({@link #saysTheConnectionIsGone}). A
     * manufactured one would put an attempt whose pooled connection is perfectly healthy into the
     * replay and call {@link #distrustPool} on it over a database that had simply refused a new
     * connection.
     * <p>
     * It buys exactly that and no more, which is worth being precise about: where the driver's own
     * refusal is of class {@code 08} - mysql answers its connection limit with {@code 08004} - the
     * attempt is classified as a dropped connection whatever this method does, the original being the
     * cause of this one and every chain of a failure being walked. What this keeps is the promise that
     * the retry changes no classification: a refusal reaches {@code write()} as the same thing it
     * reached it as before there was any retry here at all.
     * <p>
     * Which of the two bounds ended the wait is named rather than left to be guessed: the property is
     * the thing to raise only where the property is what ran out, and where the replay window of the
     * caller is the shorter one - the default has it at a sixth of the property - raising the property
     * moves nothing at all.
     */
    private static SQLTimeoutException catalogConnectTimedOut(String connectionString, long poolTimeoutSeconds,
            boolean endedByReplayWindow, long waitedMs, int attempts, SQLException last) {
        final SQLTimeoutException timeout=new SQLTimeoutException("no connection to "
            +CachedConnection.safeUrl(connectionString)+" could be opened for the tree catalog within "
            +waitedMs+"ms ("+attempts+" attempts, "+(endedByReplayWindow
                ? "what was left of the replay window of the write that asked for it, which is the shorter"
                    +" bound here: "+CachedConnection.POOL_TIMEOUT_PROPERTY+" is "+poolTimeoutSeconds+"s"
                : CachedConnection.POOL_TIMEOUT_PROPERTY+"="+poolTimeoutSeconds+"s")
            +"): the database took no connection for the moment,"
            +" last error: "+CachedConnection.redact(last.getMessage(), connectionString),
            last.getSQLState(), last.getErrorCode());
        timeout.initCause(CachedConnection.reported(last, connectionString));
        return timeout;
    }

    /**
     * One attempt of {@link #newCatalogConnection}, established and set up or left holding nothing.
     * Failures leave here as the driver reported them, checked and unchecked alike: what a retry is
     * decided on is the chain of the original, and the redaction is the caller's - a redacted copy is
     * rebuilt link by link, so redacting an attempt that is about to be retried would pay for a
     * failure nobody ever sees.
     */
    private Connection connectCatalog(String connectionString, CachedConnection.ConnectDialect dialect,
            long timeoutSeconds) throws SQLException {
        // A driver is free to write into the map it is handed, so every attempt gets one of its own.
        final Properties properties=new Properties();
        final boolean readBoundSet=dialect!=null && timeoutSeconds>0
            && dialect.bound(connectionString, properties, timeoutSeconds);
        final Connection con=DriverManager.getConnection(connectionString, properties);
        try {
            con.setAutoCommit(false);
            con.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
        }catch (SQLException | RuntimeException e) { // nothing else holds this connection yet: it would leak
            closeQuietly(con, e);
            throw e;
        }
        if (readBoundSet) {
            try {
                // only where this code set one: a read bound of the connection string is the
                // administrator's and is not lifted along with it, exactly as the pool leaves it
                con.setNetworkTimeout(Runnable::run, 0);
            }catch (SQLException | RuntimeException e) {
                // A driver that will not take the bound back leaves it in force for the life of the
                // connection, and that bound is the one this attempt was given - near the end of the
                // deadline of the retry, a second. The connection is kept all the same, which is the
                // pool's own answer to this failure: it stops pooling such a connection and still hands
                // it to the borrower that is waiting. Failing here instead would stop the backend opening
                // on a driver whose setNetworkTimeout is not implemented at all, where the pooled
                // connection beside it works - and there is no state to fail with that write() does not
                // read as a connection the database dropped. So it is reported, at the bound in force.
                logger.warn(LocalizableMessage.raw("jdbc: the catalog connection of backend %s keeps the %ds read bound its login was given, so a statement of the catalog slower than that fails on it: %s",
                    config.getBackendId(), timeoutSeconds, stackTraceToSingleLineString(e)));
            }
        }
        return con;
    }

    /** Closes a connection nothing holds yet, reporting the failure of the close on the one being unwound. */
    private static void closeQuietly(Connection con, Throwable unwinding) {
        try {
            con.close();
        }catch (SQLException | RuntimeException e) {
            // the unchecked one as well: this runs from the catch of a failure it must not replace
            // (JLS 14.20.2), which is the rule every close of this class keeps
            unwinding.addSuppressed(e);
        }
    }

    /**
     * The connection the catalog table of a backend is read and written on, and the transaction over
     * it. No other connection touches that table.
     * <p>
     * It belongs to a write transaction and not to the storage, and is opened at the first row that
     * transaction has to write - so what costs a physical connect is a write which enrols a tree the
     * catalog does not already record, and nothing else: a read-only storage opens none, a
     * transaction that opens no tree opens none, and neither does one whose trees are all recorded
     * already, which is every open after the first ({@link JDBCStorage#enrolledTrees} is the storage's and
     * outlives them). The open of a backend is therefore one connect, and so is every later write
     * that names a tree for the first time - {@code dsconfig create-backend-index} reaches exactly
     * that, opening its new tree inside a write of its own on a running server. The connect is
     * retried the way the pool retries its own for that reason; see {@link JDBCStorage#newCatalogConnection}.
     * <p>
     * Storage-scoped rather than transaction-scoped it cannot be: it is closed with the transaction
     * because the rows it writes are the transaction's, and a connection outliving them would be a
     * second pooled-connection lifetime for this class to get right.
     * <p>
     * Why the rows are not written on the caller's connection is in {@link
     * WriteableTransactionTransactionImpl#enrolInCatalog}: they have to be committed, and that commit
     * must not be the caller's. Why the read is not either is in {@link
     * WriteableTransactionTransactionImpl#readEnrolledTrees}: a select of the caller's transaction
     * would hold a lock on the catalog table for the whole life of that transaction, and the rows it
     * decides are written from here.
     */
    final class CatalogSession implements Closeable {
        private Connection con;
        private WriteableTransactionTransactionImpl txn;

        // The moment the replay window of the write() this session belongs to runs out, as
        // System.nanoTime() reads it - null where nothing above this session replays, which is the
        // importer and nothing else. Boxed rather than given a sentinel: nanoTime() is documented to
        // return an arbitrary long, so there is no reading of it that could stand for "no window".
        private Long replayWindowEndsAt;

        /**
         * Tells this session the wall-clock window the {@link JDBCStorage#write} above it bounds its
         * replay by, which the connect of the catalog may not outlast: the connect runs inside one
         * attempt of that loop, so a wait longer than what is left of the window reaches it with the
         * window already spent and is thrown unreplayed - see {@link JDBCStorage#newCatalogConnection}.
         * Called once per attempt, before the operation runs; a session nobody calls it on waits the
         * pool timeout out in full, which is what the importer does.
         */
        void boundedAlsoBy(long replayWindowEndsAtNanos) {
            replayWindowEndsAt=replayWindowEndsAtNanos;
        }

        /**
         * That window as a deadline of the clock the connect measures itself by, or {@link
         * Long#MAX_VALUE} where there is no window - the value {@link CachedConnection#deadlineOf}
         * gives a wait with no end, so that the shorter of the two is a plain {@code min}. The two
         * readings are taken here rather than one of them being carried in: a nanoTime window and a
         * currentTimeMillis deadline are two clocks, and they can only be put together at one moment.
         * A window already spent gives the moment itself, which is one attempt and then a timeout.
         */
        private long budgetDeadline() {
            if (replayWindowEndsAt==null) {
                return Long.MAX_VALUE;
            }
            final long leftNanos=replayWindowEndsAt-System.nanoTime();
            final long now=System.currentTimeMillis();
            return leftNanos<=0 ? now : now+leftNanos/1_000_000L;
        }

        /** The connection, opened at the first read or write the catalog needs and shared by the rest. */
        Connection connection() throws SQLException {
            if (con==null) {
                con=newCatalogConnection(budgetDeadline());
            }
            return con;
        }

        /** Whether this session is holding a connection already, so that a caller knows what it made. */
        boolean isEstablished() {
            return con!=null;
        }

        /**
         * A transaction over that connection, for its row statements alone: an upsert and a delete are
         * per engine, and writing the catalog through the very ones every other tree is written through
         * is what keeps its rows the same shape as theirs. It opens no tree and stamps no table, so the
         * sessions it carries of its own are never opened.
         */
        WriteableTransactionTransactionImpl transaction() throws SQLException {
            final Connection con=connection();
            if (txn==null) {
                txn=new WriteableTransactionTransactionImpl(con);
            }
            return txn;
        }

        void commit() throws SQLException {
            con.commit();
        }

        /**
         * What a failed statement left behind must not poison the write of the next row: postgres
         * refuses every further statement of a transaction whose statement failed (25P02) until it is
         * rolled back, and this connection outlives the row that failed on it.
         */
        void reset() {
            if (con!=null) {
                try {
                    con.rollback();
                }catch (SQLException | RuntimeException e) {
                    // the unchecked one as well: a driver is free to answer a rollback on a connection the
                    // database dropped with one, and this runs from the catch of a failure it must not
                    // replace - the caller goes on to report that failure, and in createCatalogTable() to
                    // tolerate a table another session created while this one was creating it
                    close();
                }
            }
        }

        /**
         * The unchecked failure of a close is taken like the checked one, and the session is given up
         * in a finally: this is called from {@link #reset}, which runs from the catch of a failure it
         * must not replace (JLS 14.20.2) - {@code createCatalogTable()} goes on from there to tolerate
         * a table another session created while this one was creating it. A session whose connection
         * would not close is left holding none rather than holding a dead one.
         * <p>
         * The catch of {@code write()}'s own finally is the same guard one layer out, kept as the
         * belt to this one's braces: it was the only guard while this method let an unchecked failure
         * past, and a session that stops swallowing must not have to be found through a failure it
         * replaced.
         */
        @Override
        public void close() {
            if (con!=null) {
                try {
                    con.close();
                }catch (SQLException | RuntimeException e) {
                    logger.trace(LocalizableMessage.raw("jdbc: unable to close the catalog connection: %s", stackTraceToSingleLineString(e)));
                }finally {
                    con=null;
                    txn=null;
                }
            }
        }
    }

    // The connection the comment statements of one sweep of openTree() calls share. Opening a
    // backend opens every tree it holds (about 25 for a stock suffix), so a connection per stamp
    // would mean that many physical connects on the first open after an upgrade - the one open
@@ -1134,21 +1632,32 @@
            if (con!=null) {
                try {
                    con.rollback();
                }catch (SQLException e) {
                }catch (SQLException | RuntimeException e) {
                    // the unchecked one as well: a driver is free to answer a rollback on a connection the
                    // database dropped with one, and this runs from the catch of a failure it must not
                    // replace - the caller goes on to report that failure, and in createCatalogTable() to
                    // tolerate a table another session created while this one was creating it
                    close();
                }
            }
        }

        /**
         * The unchecked failure of a close is taken like the checked one, and the session is given up
         * in a finally, for the reason {@link CatalogSession#close} gives: this runs from {@link
         * #reset}, which runs from the catch of a failure it must not replace - and a stamp that
         * failed must never become the outcome of the open it was issued from.
         */
        @Override
        public void close() {
            if (con!=null) {
                try {
                    con.close();
                }catch (SQLException e) {
                }catch (SQLException | RuntimeException e) {
                    logger.trace(LocalizableMessage.raw("jdbc: unable to close the comment connection: %s", stackTraceToSingleLineString(e)));
                }finally {
                    con=null;
                }
                con=null;
            }
            mysqlBackslashEscape=null; // it described the session that has just gone
        }
@@ -1337,8 +1846,10 @@
    }

    // Returns the comment currently stored on the table, or null when there is none. The dialect is
    // passed in rather than read off the connection: this runs on the stamp connection, which is
    // not a pooled one, and only for the dialects commentTable() recognizes.
    // passed in rather than read off the connection: the stamp sweep runs this on a connection of its
    // own, a clear runs it on the pooled one it did its work on, and both only for the dialects
    // commentTable() recognizes. It is a read and nothing else, and CachedConnection.close() rolls
    // back before the connection is handed on, so a clear leaves no transaction of its own behind.
    String readStoredComment(Connection con, Dialect dialect, String tableName) throws SQLException {
        final String sql;
        final String arg;
@@ -1353,7 +1864,7 @@
            break;
        case ORACLE:
            sql="select comments from user_tab_comments where table_name=?";
            arg=tableName.toUpperCase();
            arg=tableName.toUpperCase(Locale.ROOT);
            break;
        case MICROSOFT:
            sql="select cast(value as nvarchar(4000)) from sys.extended_properties where class=1 and major_id=object_id(?) and minor_id=0 and name='MS_Description'";
@@ -1415,7 +1926,7 @@
                    break;
                case ORACLE:
                    sql="begin dbms_stats.gather_table_stats(user, ?); end;";
                    args=new String[]{tableName.toUpperCase()};
                    args=new String[]{tableName.toUpperCase(Locale.ROOT)};
                    break;
                case MICROSOFT:
                    sql="update statistics "+tableName;
@@ -1469,6 +1980,41 @@
        return allRefreshed;
    }

    /**
     * Whether a table of this name is one the given connection reaches: in its database, and in one of
     * the schemas an unqualified name of it resolves in - see {@link TableScope}, which is where the
     * reason for each half of that question is. Asked of the catalog by name rather than by listing
     * every table of the database: openTree(createOnDemand) asks it for every tree of the backend -
     * about 25 of them for a stock suffix - on every open, on a database this backend may well be
     * sharing with something else.
     */
    boolean isExistsTable(Connection con, TableScope scope, String tableName) {
        // bounded as the operation it is, not as the bulk statement it guards and not as the class of
        // the transaction that happens to ask (#882): it reads a data dictionary rather than the data,
        // so a wait here is the metadata lock of another session
        try {
            return bounded(con, StatementBound.OPERATION, () -> {
                final DatabaseMetaData metaData = con.getMetaData();
                // asked with no schema pattern and read through the scope instead: what an unqualified
                // statement reaches is a path of schemas and not one of them, and a pattern is no way to
                // name a path - nor an exact way to name even one of it, "_" being a wildcard there
                try (final ResultSet rs = metaData.getTables(scope.catalog, null,
                        storedIdentifier(metaData, tableName), new String[]{"TABLE"})) {
                    while (rs.next()) {
                        // the name still has to be compared: "_" is a single-character wildcard in a
                        // metadata pattern, so "opendj_<hash>" also matches a table named "opendjX<hash>"
                        if (tableName.equalsIgnoreCase(rs.getString("TABLE_NAME")) && scope.covers(rs)) {
                            return true;
                        }
                    }
                }
                return false;
            });
        } catch (Exception e) {
            throw new StorageRuntimeException(e);
        }
    }

    @Override
    public void removeStorageFiles() throws StorageRuntimeException {
        final boolean isOpen=getStorageStatus().isWorking();
@@ -1479,36 +2025,670 @@
                throw new StorageRuntimeException(e);
            }
        }
        final Set<TreeName> trees=listTrees();
        if (!trees.isEmpty()) {
            try (final Connection con = getValidatedConnection()) {
                try {
                    for (final TreeName treeName : trees) {
                        try (final PreparedStatement statement = con.prepareStatement("drop table " + getTableName(treeName))) {
                            execute(statement, StatementBound.BULK);
        try (final Connection con = getValidatedConnection()) {
            // where an unqualified name of this connection resolves, which every lookup below is
            // narrowed to: the skip in the loop decides between leaving a row where it is and dropping
            // the table it names, and a table of that name in another database of the server must not
            // be allowed to answer for this one - nor a table of this backend go unfound for living in
            // another schema of the search path than the one the connection works in
            final TableScope scope=TableScope.of(this, con);
            // the catalog names what this backend owns, and only that: listTrees() also names the
            // shared compressed schema trees, which another backend of this database may be the only
            // owner of and which a clear must therefore leave exactly where they lie (#881)
            final List<String> skippedRows=new ArrayList<>(); // rows the read could not act on: reported below
            final Map<TreeName,String> trees=catalogTables(con, scope, skippedRows);
            final TreeName catalogTree=getCatalogTree();
            int dropped=0;
            // the same count with the catalog itself left out, which is what says whether this clear
            // removed anything of the backend: a catalog table standing over rows that name nothing -
            // a backup restored older than the tables it was taken beside - is dropped like any other
            // and would otherwise make a clear that removed no tree at all look like a clear that did
            // something. See reportClearOutcome()
            int droppedTrees=0;
            // counted without the catalog, for the reason droppedTrees is kept apart from dropped: the
            // catalog is walked by this loop like any other table, so a catalog table that went between
            // the lookup of catalogTables() and the loop's own would otherwise be summed up as a tree of
            // this backend that had lost its table
            int missingTrees=0;
            try {
                for (final Map.Entry<TreeName,String> tree : trees.entrySet()) {
                    final String tableName=tree.getValue();
                    final boolean isCatalog=catalogTree.equals(tree.getKey());
                    if (!isExistsTable(con, scope, tableName)) { // a row of the catalog outliving its table
                        reportClearLine(LocalizableMessage.raw(
                            "jdbc: backend %s names tree %s, whose table %s is not there: nothing to drop for it",
                            config.getBackendId(), tree.getKey(), tableName));
                        if (!isCatalog) {
                            missingTrees++;
                        }
                        continue;
                    }
                    con.commit();
                } catch (SQLException e) {
                    dropTable(con, tableName);
                    dropped++;
                    if (!isCatalog) {
                        droppedTrees++;
                    }
                }
                con.commit();
            } catch (Exception e) {
                // every failure of the loop and not the SQLException alone: the lookup deciding each
                // drop answers with a StorageRuntimeException of its own, and a drop left pending by one
                // of those has to go back here rather than wait for the connection to be handed back
                try {
                    con.rollback();
                } catch (SQLException e2) {}
                throw e instanceof StorageRuntimeException ? (StorageRuntimeException) e : new StorageRuntimeException(e);
            }
            // all tables are gone: a table recreated later deserves a fresh stamp attempt, and the
            // memoized table name of a tree nothing holds any more is of no use to anyone
            for (final TreeName treeName : trees.keySet()) {
                tree2table.invalidate(treeName);
                unstampableTrees.remove(treeName);
            }
            try {
                reportClearOutcome(con, scope, dropped, droppedTrees, missingTrees, skippedRows);
            } catch (RuntimeException e) {
                // the clear itself is done and committed: an account of what it left standing must not be
                // the thing that reports it as failed, and a caller retrying it would find nothing to drop
                logger.trace(LocalizableMessage.raw("jdbc: unable to report what the clear left standing: %s",
                    stackTraceToSingleLineString(e)));
            }
        } catch (StorageRuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw new StorageRuntimeException(e);
        } finally {
            // the catalog went with the rest: the next tree enrolled creates its table again. The
            // online import needs exactly that - the storage which has just dropped its tables is the
            // one going on to open a root container and enrol every tree of it anew, which is what the
            // forgotten enrolments make it do rather than skip as already recorded.
            catalogTableOpened=false;
            enrolledTrees.clear();
            if (!isOpen) {
                close();
            }
        }
    }

    /**
     * Drops one table of a clear. It is a method of its own so that the order {@link
     * #removeStorageFiles()} drops in can be watched from a test: what names the trees has to outlive
     * them, and that guarantee is the loop's - it holds because the loop walks the catalog's map in
     * the order that map was built in, and a test asserting on the map instead would go on passing
     * over a loop that had stopped doing so.
     */
    void dropTable(Connection con, String tableName) throws SQLException {
        try (final PreparedStatement statement = con.prepareStatement("drop table " + tableName)) {
            // bulk, as #882 made every drop of this backend: nobody waits on a clear, and what it takes
            // follows the size of the table rather than the work of a caller
            execute(statement, StatementBound.BULK);
        }
    }

    /**
     * Where a line of the account a clear gives of itself goes: the logger, and nothing else in
     * production. It is a method of its own so that a case can hold that account to what it says -
     * these lines change no state at all, so every assertion a clear can be given about the database
     * passes just as well with all of them deleted, and this report has now been changed in three
     * rounds of review with nothing able to fail. See {@code TestCase.ReportingStorage}, which
     * collects them.
     */
    void reportClearLine(LocalizableMessage line) {
        logger.warn(line);
    }

    /**
     * Reports what a clear did not remove, once everything the catalog named is gone.
     * <p>
     * An "opendj" table still standing at that point is named by no catalog of this backend, and its
     * name says nothing about whose it is - a table is named after the hash of its tree name. What
     * does say so is the comment a table is stamped with as it is opened (#866): the tree name in
     * plain text. A table whose stamp names a tree of a base DN this backend does not serve belongs to
     * a backend sharing this database (#873) and is passed over in silence; one whose stamp names a
     * tree of this backend is reported as its own, and so as removable by hand; one carrying no stamp
     * at all - left by a version stamping no table, or by a database that refused the comment - can be
     * attributed to nobody and is reported as exactly that. A stamp the database would not give up is
     * reported apart from all of these: it says nothing either way, and counting it as a table without
     * a stamp would turn a connection that died halfway into a confident line about tables this
     * backend may well own.
     * <p>
     * The silence has a cost worth stating: a table stamped with a tree of a base DN that was taken
     * out of the configuration while the backend was disabled reads exactly like a table of a backend
     * sharing the database, the stamp naming the tree and never the backend it belonged to, so it is
     * passed over too. What is left of such a base DN is found by its stamp and removed by hand.
     * <p>
     * The shared compressed schema pair is left out of all of it: it is kept on purpose (#881), so it
     * is no leftover of anything, and naming it here would be asking for the removal of the one thing
     * this code goes out of its way to spare.
     * <p>
     * A clear which removed no tree of this backend is called out ahead of all of it: #888 was exactly
     * such a clear, and it went by without a word in the log. A backend upgraded in place is the one
     * case where a clear drops nothing while there is something to drop - nothing enrols a tree before
     * {@link #removeStorageFiles()} runs, so the first offline clear of such a backend finds no
     * catalog at all - and the line says so rather than leaving it to be found out.
     * <p>
     * The catalog table is no term of that count. It is dropped like any other and by the same loop,
     * so a catalog standing over rows that name nothing - a backup restored older than the tables it
     * was taken beside - is one table dropped and not one tree removed, and the line has to fire there
     * too: what an operator meets in that case is the same clear that removed none of their data.
     * <p>
     * A database which would not say what is standing gets a line of its own, whatever the clear
     * dropped. What was left behind is exactly what could not be found out there, so it is no more a
     * clear that left nothing than one that left something, and the count of what it did drop is the
     * only thing that can still be stated: reporting it through the line above would say "the clear
     * dropped no table at all" of a clear that dropped a dozen.
     * <p>
     * A row of the catalog the read passed over is reported wherever the clear got to, that line
     * depending on nothing this database was asked afterwards: what such a row records is outside the
     * namespace {@link #leftoverTables} scans, so no other line here can name it. The row itself does
     * not survive the clear - the catalog names itself last and the loop drops that table with every
     * row still in it - which is why the line is the only surviving copy of what the row said, and
     * why it names what the row recorded rather than telling an operator to go and look. Nothing this
     * version writes makes such a row - {@link #getTableName} names every table {@code opendj_<hash>}
     * - so it is the account of a database written into by something else.
     * <p>
     * It is a term of the "dropped nothing" line all the same, and for one state only: a catalog whose
     * table is there names itself, so a clear reading any row at all normally drops that one and the
     * term is carried by the drop count beside it. Where it is not is where the catalog table went
     * between the read of its rows and the loop that drops them - another process clearing the same
     * backend - and there the clear has read a row, dropped nothing, and has this row as the whole of
     * what it can say. Without the term it says nothing at all, which is the silence of #888.
     */
    void reportClearOutcome(Connection con, TableScope scope, int dropped, int droppedTrees, int missingTrees,
            List<String> skippedRows) {
        final ClearLeftovers leftovers=leftoverTables(con, scope);
        if (leftovers==null) {
            // a line of its own and not a clause of the one below: this says nothing about whether
            // anything was left behind, so a clear that dropped its tables must not be reported here as
            // one that dropped none - and one that dropped none must still say so, that silence being
            // the whole of #888
            reportClearLine(LocalizableMessage.raw("jdbc: backend %s: %s (it dropped %d table(s) in all, its own catalog among them where there was one) and %d of the trees its catalog names had lost their table already; what else is standing could not be read off this database, so this clear says nothing about it.%s",
                config.getBackendId(), removedTrees(droppedTrees), dropped, missingTrees,
                droppedTrees==0 ? " "+CLEAR_DROPPED_NOTHING : ""));
            reportSkippedRows(skippedRows); // read off the catalog and not off this database: still worth stating
            return;
        }
        final int ours=leftovers.ours.size();
        final int unattributed=leftovers.unattributed.size();
        final int unreadable=leftovers.unreadable.size();
        // first of the lines, and not last: on a backend upgraded in place every table of it is
        // unstamped and lands in the list below, and the operator has to be told why before being
        // handed a list of tables their own backend is very probably still using.
        // Decided on the drops of trees and not on every drop: the catalog table is dropped by the same
        // loop, so a catalog standing over rows that name nothing makes "dropped" one while no tree of
        // this backend was removed - which is the state this line exists to explain.
        // Each tree which had lost its table is logged as the loop skips it; this line only sums them up.
        // "there was something to act on" and not "something is still standing": a clear which dropped
        // its own catalog and removed no tree of the backend is the #888 outcome exactly, and it says so
        // whether or not the scan afterwards found anything to attribute. Without the two terms on the
        // right the line is silent in that case while the same clear on a database whose listing failed
        // announces itself - the same clear, told two ways.
        // The last of them is not spare: a clear normally drops the catalog table it read its rows out
        // of, so a passed-over row comes with a drop - except where that table went while this clear was
        // running, which is a clear that read a row, dropped nothing, and has that row as all it can say
        if (droppedTrees==0 && (missingTrees>0 || ours>0 || unattributed>0 || unreadable>0
                || dropped>0 || !skippedRows.isEmpty())) {
            reportClearLine(LocalizableMessage.raw("jdbc: backend %s: %s (it dropped %d table(s) in all, its own catalog among them where there was one): %d of the trees its catalog names had lost their table already, and %d table(s) of this backend were named by no catalog, %d could not be attributed to anyone and %d could not be read. %s",
                config.getBackendId(), removedTrees(droppedTrees), dropped, missingTrees, ours, unattributed,
                unreadable, CLEAR_DROPPED_NOTHING));
        }
        reportSkippedRows(skippedRows); // after the reason above and among the lists, being a list itself
        if (ours>0) {
            reportClearLine(LocalizableMessage.raw("jdbc: backend %s: %d table(s) of %s hold trees of this backend that its catalog does not name, and the clear left them where they are: %s. A tree is enrolled as it is opened read-write and by no other means, so such a table is one of a tree of a base DN this backend still serves that was taken out of the configuration while it was disabled - an attribute index, say - or one left by a version keeping no catalog: it is this backend's own and can be removed by hand, and re-adding the tree it belongs to adopts it with the rows it still holds",
                config.getBackendId(), ours, scope.name(), leftovers.ours));
        }
        if (unattributed>0) {
            reportClearLine(LocalizableMessage.raw("jdbc: backend %s: %d opendj table(s) of %s are named by no catalog of this backend and carry no tree stamp, so nothing says whose they are: %s. They may hold the trees of a backend sharing this database, which nothing forbids, or be leftovers of a version stamping no table at all - a table is named after the hash of its tree name and can be attributed by no other means. They were left exactly where they are",
                config.getBackendId(), unattributed, scope.name(), leftovers.unattributed));
        }
        if (unreadable>0) {
            reportClearLine(LocalizableMessage.raw("jdbc: backend %s: the stamp of %d opendj table(s) of %s could not be read, so this clear says nothing about whose they are: %s. They were left exactly where they are",
                config.getBackendId(), unreadable, scope.name(), leftovers.unreadable));
        }
    }

    /**
     * What a clear did to the trees of its backend, in the one wording both lines of the report use:
     * a condition an operator greps for - and a case asserts on - must not be phrased one way where
     * the leftover scan answered and another way where it did not.
     */
    private static String removedTrees(int droppedTrees) {
        return droppedTrees==0 ? "the clear removed no tree of this backend"
            : "the clear removed "+droppedTrees+" tree(s) of this backend";
    }

    /**
     * Reports the rows of the catalog the clear could not act on; see {@link #readCatalogRows} for
     * what makes a row one of these and {@link #reportClearOutcome} for why they are a line of their
     * own. Silent where there are none, which is every clear of a catalog this backend wrote.
     * <p>
     * The row is gone by the time this prints and what it recorded is not: the catalog names itself
     * last, so the loop drops the table holding these rows along with every other - and where that
     * table went on its own between the two lookups, it took them with it just the same. That is what
     * the line has to say, and why it carries the recorded name rather than sending an operator to a
     * table that is no longer there.
     */
    private void reportSkippedRows(List<String> skippedRows) {
        if (skippedRows.isEmpty()) {
            return;
        }
        reportClearLine(LocalizableMessage.raw("jdbc: backend %s: %d row(s) of its catalog named nothing this clear could drop and were passed over: %s. The rows are gone with the catalog table, which a clear drops last; whatever they record was left standing, and no other line of this clear names it: the tables of this backend are named after the hash of a tree name, so what such a row records is outside the names a clear can account for. This line is the only surviving copy of it. A catalog holding such a row was written into by something other than this backend",
            config.getBackendId(), skippedRows.size(), skippedRows));
    }

    /**
     * Why a clear can remove no tree while there is something to remove, said wherever one did: it is
     * the silence of #888, and the one thing an operator reading such a line has to be told.
     */
    private static final String CLEAR_DROPPED_NOTHING="A backend upgraded from a version keeping no catalog has to be started once before its first offline \"import-ldif --clearBackend\": nothing enrols a tree before the clear runs, so that first clear finds a catalog that is not there - or, where the tables were restored from a backup taken beside an older one, a catalog that is there and names nothing - and removes no tree either way";

    /** What a clear left standing, told apart by the tree stamp of each table; see {@link #reportClearOutcome}. */
    static final class ClearLeftovers {
        /** Tables whose stamp names a tree of this backend: its own, and removable by hand. */
        final List<String> ours=new ArrayList<>();
        /** Tables carrying no stamp naming a tree: they can be attributed to nobody. */
        final List<String> unattributed=new ArrayList<>();
        /** Tables whose stamp the database would not give up: they are attributed neither way. */
        final List<String> unreadable=new ArrayList<>();
    }

    /**
     * The "opendj" tables this connection reaches - see {@link TableScope} - that this backend can say
     * something about, or {@code null} where the database would not list them. A table stamped with a
     * tree this backend does not serve is in none of the lists: it is a backend sharing this database
     * (#873) that it belongs to, and no part of this clear's outcome.
     */
    ClearLeftovers leftoverTables(Connection con, TableScope scope) {
        // the shared compressed schema pair is left standing on purpose, so it is no leftover of
        // anything and reporting it would be pointing at the one thing this code goes out of its way
        // to keep. Taken out by name and not by stamp: an installation may hold the pair unstamped,
        // from a version that commented no table at all.
        final Set<String> leftOnPurpose=new HashSet<>();
        for (final TreeName treeName : SHARED_COMPRESSED_SCHEMA_TREES) {
            leftOnPurpose.add(readTableName(treeName).toLowerCase(Locale.ROOT));
        }
        final ClearLeftovers leftovers=new ClearLeftovers();
        try {
            // by name and not by row: the listing is asked with no schema pattern and read back through
            // the resolution path (see TableScope), so two schemas of that path holding a table of the
            // same name both answer here - and there is exactly one thing this report can say of that
            // name, the stamp being read through an unqualified statement that resolves whichever of
            // them the path reaches first. Named twice it would read as two leftovers where the clear
            // can account for one. By the name exactly as the database spells it, and not folded: what
            // this collapses is one name in two schemas, while two names differing only in case are
            // two tables of a case-preserving engine and each is a leftover of its own
            final Set<String> standing=new LinkedHashSet<>();
            final DatabaseMetaData metaData=con.getMetaData();
            try (final ResultSet rs=metaData.getTables(scope.catalog, null,
                    storedIdentifier(metaData, "opendj%"), new String[]{"TABLE"})) {
                while (rs.next()) {
                    final String tableName=rs.getString("TABLE_NAME");
                    if (tableName==null) { // a row naming no table names nothing this clear can report
                        continue;
                    }
                    if (!leftOnPurpose.contains(tableName.toLowerCase(Locale.ROOT)) && scope.covers(rs)) {
                        standing.add(tableName);
                    }
                }
            }
            // the stamps are read once the metadata result set is closed: they are queries of this very
            // connection, and a driver may hold it for the whole of that result set
            final Dialect dialect=dialectOf(con);
            if (dialect==null) {
                // no comment readback is known for this engine, so no table of it can be attributed to
                // anyone at all. That is a different thing from a table which carries no stamp, and saying
                // the second would be telling an operator that every table of every backend of this
                // database is of unknown ownership when the truth is that nothing was ever asked
                leftovers.unreadable.addAll(standing);
                return leftovers;
            }
            for (final String tableName : standing) {
                final TreeName stamp;
                try {
                    stamp=stampedTree(con, dialect, tableName);
                } catch (SQLException | RuntimeException e) {
                    // this table alone is unaccounted for, and the ones after it need not be: postgres
                    // refuses every further statement of a transaction whose statement failed (25P02), so
                    // the read that failed is rolled back before the next table is asked about. There is
                    // nothing pending to lose - the clear committed its drops before this ran
                    logger.trace(LocalizableMessage.raw("jdbc: unable to read the stamp of table %s: %s",
                        tableName, stackTraceToSingleLineString(e)));
                    leftovers.unreadable.add(tableName);
                    try {
                        con.rollback();
                    } catch (SQLException e2) {}
                    throw new StorageRuntimeException(e);
                    continue;
                }
            } catch (Exception e) {
                throw new StorageRuntimeException(e);
                if (stamp==null) {
                    leftovers.unattributed.add(tableName);
                } else if (isOwnTree(stamp)) {
                    leftovers.ours.add(tableName+" ("+stamp+")");
                }
            }
            // all tables are gone: forget the mappings so listTrees() consumers skip the dropped trees
            for (final TreeName treeName : trees) {
                tree2table.invalidate(treeName);
                unstampableTrees.remove(treeName); // a table recreated later deserves a fresh stamp attempt
            }
        } catch (SQLException e) {
            logger.trace(LocalizableMessage.raw("jdbc: unable to look for the tables a clear left behind: %s",
                stackTraceToSingleLineString(e)));
            return null;
        }
        if (!isOpen) {
            close();
        return leftovers;
    }

    /**
     * The tree named by the comment this table carries (#866), or {@code null} where it carries none
     * or where what it carries is not the name of a tree. The stamp is the only thing that attributes
     * a table to a backend at all - a table name is a bare hash - and stamping is best-effort, so the
     * absence of one states nothing.
     * <p>
     * A read the database refused is passed to the caller rather than answered as an absent stamp: the
     * two say different things, and the second would let a connection that died halfway be reported as
     * a row of tables nothing can be said about. An engine with no readback of its own is the same
     * distinction one step earlier, and is answered by the caller: it puts every table of such an
     * engine where nothing was asked of it belongs, which is not where a table without a stamp goes.
     */
    private TreeName stampedTree(Connection con, Dialect dialect, String tableName) throws SQLException {
        final String comment=readStoredComment(con, dialect, tableName);
        if (comment==null || comment.isEmpty()) {
            return null;
        }
        try {
            return TreeName.valueOf(comment);
        } catch (RuntimeException e) { // a comment of somebody else's making: no stamp of this backend's kind
            return null;
        }
    }
	

    /**
     * The base DN the compressed schema trees of this backend are named under since #881, spelled out
     * here for the reason {@link #SHARED_COMPRESSED_SCHEMA_TREES} is: the prefix is built by a private
     * method of {@code PersistentCompressedSchema}, escapes and all. A table stamped with one of these
     * carries this backend's id in plain text, so a clear that finds one standing can say whose it is.
     */
    private String ownCompressedSchemaBaseDN() {
        return SHARED_COMPRESSED_SCHEMA_BASE_DN+"_"+escapedBackendId();
    }

    /**
     * The backend id as one component of a tree name. A tree name is {@code /<base DN>/<id>} and is
     * read back by splitting on its slashes ({@code TreeName.valueOf}), so an id carrying one of them
     * would name a tree that parses into another tree than it was built from - and a table is stamped
     * with that name (#866), so a clear reading the stamp of a table of this backend's own would then
     * fail to recognize it and pass it over in silence. The escape is the one {@code
     * PersistentCompressedSchema} spells its own prefix with, percent first so that the escape of the
     * slash cannot be produced twice, and it leaves an id of the ordinary shape exactly as it is -
     * which is what keeps the table names of an installation unchanged.
     */
    private String escapedBackendId() {
        return config.getBackendId().replace("%", "%25").replace("/", "%2F");
    }

    /**
     * Whether this tree is one of this backend's own: a tree of a base DN it serves, its own catalog,
     * or its own pair of compressed schema trees. The catalog counts because a clear drops it last, so
     * one still standing is a clear of this backend that did not get to the end, and never anything of
     * anybody else's. The compressed schema pair counts because since #881 it is named after the
     * backend id (#873) and so belongs to this backend as plainly as any tree of a base DN it serves -
     * where the legacy pair, named from a literal, belongs to no backend in particular and is reported
     * by nobody.
     */
    private boolean isOwnTree(TreeName treeName) {
        if (getCatalogTree().equals(treeName) || ownCompressedSchemaBaseDN().equals(treeName.getBaseDN())) {
            return true;
        }
        final SortedSet<DN> baseDNs=config.getBaseDN();
        if (baseDNs==null) {
            return false;
        }
        for (final DN baseDN : baseDNs) {
            // every tree of an entry container is named after the normalized form of its base DN,
            // which is what EntryContainer builds its tree names from
            if (treeName.getBaseDN().equals(baseDN.toNormalizedUrlSafeString())) {
                return true;
            }
        }
        return false;
    }

    /**
     * The database and the schemas a connection reaches with an unqualified name: what every table
     * lookup of this backend is narrowed to.
     * <p>
     * The database is the half that has to narrow. Asked with a null catalog the question spans the
     * whole server on some drivers - Connector/J reads a null catalog as "any database" since 8.0, and
     * its databaseTerm being CATALOG it ignores the schema pattern besides - and every answer of such a
     * lookup decides something a table of the same name in another database must have no say in. A
     * clear skips the row of a table that is gone so that it can go on, and a foreign table answering
     * for it turns that skip into an unqualified "drop table" of a table that is not in this database,
     * failing the clear on this attempt and on every attempt after it. An open of a tree creates its
     * table where there is none, and a foreign table answering for it skips the creation, leaving the
     * catalog naming a tree whose table is not here. Two backends of the stock backend id in two
     * databases of one server name their tables alike, so this is the ordinary layout and not a corner
     * of one.
     * <p>
     * The schema is the half that must not narrow to one name. The statements this scope guards are
     * unqualified, and an unqualified name resolves across a path of schemas: the whole
     * {@code search_path} on postgresql, the default schema of the user and then {@code dbo} on sql
     * server. A lookup narrowed to {@code current_schema()} alone would be the stricter question of the
     * two - an installation whose tables were created in {@code public} while the connection now works
     * in a schema of its own reads and writes them unqualified all the same, and asking only about that
     * schema would report them absent: the clear would drop nothing, which is #888 over again, and the
     * next open would create a second, empty set of tables shadowing the populated ones for every later
     * unqualified reference. The path is asked of the connection, so that a lookup answers for exactly
     * the tables the statements behind it reach - no more and no fewer.
     */
    static final class TableScope {
        /** The database of the connection, or {@code null} where the driver names none - oracle has none. */
        final String catalog;
        /**
         * The schemas an unqualified name of this connection resolves in, nearest first, or {@code null}
         * where the schema is no dimension of this engine - mysql, whose schema is its database - or
         * where the connection would not say. A null path narrows nothing, which is the question this
         * class asked before there was anything to narrow it by.
         */
        final List<String> schemas;
        /**
         * Whether the connection answered both questions. One it would not answer leaves the lookup as
         * wide as it ever was - fail-open, which is the safe direction for the schema and the weak one
         * for the database - so the caller asks again rather than latching that answer for the life of a
         * transaction; see {@link ReadableTransactionImpl#takeTableScope()}.
         */
        final boolean answered;

        private TableScope(String catalog, List<String> schemas, boolean answered) {
            this.catalog=catalog;
            this.schemas=schemas;
            this.answered=answered;
        }

        /**
         * What this connection says about where an unqualified name of it resolves. The storage is
         * taken because one engine is asked with a statement rather than with a method of its driver,
         * and a statement of this backend takes the bound of its class (#882).
         */
        static TableScope of(JDBCStorage storage, Connection con) {
            return of(storage, con, true);
        }

        /**
         * The same, told to keep quiet about a connection that will not answer. A transaction asks
         * again for as long as it is refused - a lookup left as wide as the whole server decides a
         * create and a drop - and one refusal per tree of the backend is one line per tree in the log,
         * each with a stack trace, for a thing that was already said.
         */
        static TableScope of(JDBCStorage storage, Connection con, boolean report) {
            String catalog=null;
            List<String> schemas=null;
            boolean answered=true;
            try {
                // an empty name is not the name of a database but a driver's way of saying it has none,
                // and passed to a metadata pattern it means "tables that belong to no catalog" - which is
                // not the same question and would answer nothing
                catalog=emptyToNull(con.getCatalog());
            } catch (Exception e) {
                // said out loud rather than swallowed: this decides a create and a drop, and a lookup
                // that silently reverts to the whole server is the one failure of the two that cannot be
                // seen from its outcome
                answered=false;
                log(report, "jdbc: this connection would not name the database it works in, so a table of another database of this server may answer for one of this backend's: %s", e);
            }
            try {
                schemas=schemaPathOf(storage, con);
            } catch (Exception e) {
                answered=false;
                log(report, "jdbc: this connection would not name the schemas an unqualified name of it resolves in, so a table of any schema may answer for one of this backend's: %s", e);
            }
            return new TableScope(catalog, schemas, answered);
        }

        /** Said once where it is worth saying, and kept for the trace where it would be said again. */
        private static void log(boolean report, String message, Exception e) {
            if (report) {
                logger.warn(LocalizableMessage.raw(message, stackTraceToSingleLineString(e)));
            } else {
                logger.trace(LocalizableMessage.raw(message, stackTraceToSingleLineString(e)));
            }
        }

        /** The schemas an unqualified name resolves in, in the order this engine resolves them. */
        private static List<String> schemaPathOf(JDBCStorage storage, Connection con) throws SQLException {
            final String driverName=driverNameOf(con);
            if (driverName.contains("mysql")) {
                // the schema of Connector/J is the database, and which of the two names it answers with is
                // the databaseTerm of the connection: with CATALOG - the default - getSchema() answers null
                // and the catalog above is the narrowing, and with SCHEMA it is the other way round. Asked
                // rather than assumed, so that neither setting leaves this lookup narrowed by nothing at all
                final String database=emptyToNull(con.getSchema());
                return database==null ? null : Collections.singletonList(database);
            }
            if (driverName.contains("postgres")) {
                // getSchema() is "select current_schema()" on pgjdbc - the first existing schema of the
                // search_path - while an unqualified reference resolves across the whole of it.
                // Behind a savepoint, because this runs on the caller's transaction and postgres refuses
                // every further statement of a transaction whose statement failed (25P02): a query this
                // engine turns out not to have - pgjdbc talks to more than one of them - would otherwise
                // surface as the next statement of the caller failing, with the cause nowhere near it
                final Savepoint before=savepoint(con);
                try (final PreparedStatement statement=con.prepareStatement("select unnest(current_schemas(true))")) {
                    // bounded like every other statement of this backend (#882), and by the class the lookups
                    // this scope narrows take: it reads a session setting rather than the data, and what the
                    // savepoint and the fallback below answer for is a query this engine refuses - not one it
                    // never answers at all, which is a wait holding the open of a tree with nothing to end it
                    final List<String> path=storage.executeResultSet(statement, StatementBound.OPERATION, rs -> {
                        final List<String> read=new ArrayList<>();
                        while (rs.next()) {
                            final String schema=emptyToNull(rs.getString(1));
                            if (schema!=null) {
                                read.add(schema);
                            }
                        }
                        return read;
                    });
                    release(con, before);
                    if (!path.isEmpty()) {
                        return Collections.unmodifiableList(path);
                    }
                } catch (Exception e) { // asked of getSchema() below instead, as well as it can say it
                    undo(con, before);
                    logger.debug(LocalizableMessage.raw("jdbc: unable to read the search path of this connection, which is asked for its current schema instead: %s",
                        stackTraceToSingleLineString(e)));
                }
            }
            final String schema=emptyToNull(con.getSchema());
            if (schema==null) {
                return null;
            }
            if (driverName.contains("microsoft")) {
                // an unqualified name resolves in the default schema of the user and then in dbo
                return Collections.unmodifiableList(Arrays.asList(schema, "dbo"));
            }
            // oracle resolves in the current schema, and past it through synonyms this cannot enumerate:
            // a table reached through one is not found here, and an open creates it again in the schema
            return Collections.singletonList(schema);
        }

        /**
         * A point to put a transaction back to, or {@code null} where this connection is in no
         * transaction to speak of or would not take one. A read of the search path is answered by the
         * connection of whoever asked for the scope, and a failed statement of it is theirs to be
         * spared.
         */
        private static Savepoint savepoint(Connection con) {
            try {
                return con.getAutoCommit() ? null : con.setSavepoint("opendj_search_path");
            } catch (SQLException | RuntimeException e) {
                return null;
            }
        }

        /** Puts the transaction back to where the probe found it, so that its failure stays the probe's. */
        private static void undo(Connection con, Savepoint savepoint) {
            if (savepoint!=null) {
                try {
                    con.rollback(savepoint);
                } catch (SQLException | RuntimeException e) {}
            }
        }

        /** Gives up a savepoint nothing needs any more: a transaction keeps them all until it ends. */
        private static void release(Connection con, Savepoint savepoint) {
            if (savepoint!=null) {
                try {
                    con.releaseSavepoint(savepoint);
                } catch (SQLException | RuntimeException e) {}
            }
        }

        /**
         * Whether the table this row of a listing describes is one this connection reaches. The metadata
         * pattern alone does not settle it: a schema reaches {@link DatabaseMetaData#getTables} as a
         * pattern, where "_" is a single-character wildcard, so a listing narrowed to a schema named
         * "app_data" is answered for by one named "appXdata" as well. The listings of this class are
         * asked with no schema pattern at all - a path is more than one name anyway - and read through
         * this instead.
         */
        boolean covers(ResultSet rs) throws SQLException {
            return isSameCatalog(rs.getString("TABLE_CAT")) && isOnSchemaPath(rs.getString("TABLE_SCHEM"));
        }

        /**
         * Whether the database of a listed table rules it out. A name neither side gives is no
         * narrowing: a driver naming no catalog of its own - oracle has none - must not be read as
         * naming another.
         */
        private boolean isSameCatalog(String ofTable) {
            return catalog==null || ofTable==null || ofTable.isEmpty() || catalog.equalsIgnoreCase(ofTable);
        }

        /** Whether a listed table is in one of the schemas an unqualified name of this connection resolves in. */
        private boolean isOnSchemaPath(String ofTable) {
            if (schemas==null || schemas.isEmpty() || ofTable==null || ofTable.isEmpty()) {
                return true;
            }
            for (final String schema : schemas) {
                if (schema.equalsIgnoreCase(ofTable)) {
                    return true;
                }
            }
            return false;
        }

        /** How the database and the schemas a table count was taken over are named in a log line. */
        String name() {
            final String where=schemas==null || schemas.isEmpty() ? null : String.join(", ", schemas);
            if (catalog!=null && where!=null) {
                return catalog+"."+where;
            }
            if (catalog!=null) {
                return catalog;
            }
            return where!=null ? where : "this connection";
        }

        /** An empty name is the name of nothing: see {@link #of}. */
        private static String emptyToNull(String name) {
            return name==null || name.isEmpty() ? null : name;
        }
    }

    //operation
    /**
     * {@inheritDoc}
@@ -1601,6 +2781,11 @@
            try (con) {
                driver=driverNameOf(con);
                final WriteableTransactionTransactionImpl txn=new WriteableTransactionTransactionImpl(con);
                //the connect of the catalog is made inside this attempt and retries the way a borrow does,
                //up to the pool timeout - six times this window at the defaults. Left to its own deadline
                //it would spend a window it does not own and hand the loop a failure it has classified as
                //replayable with nothing left to replay it in, so it is told where the window ends
                txn.catalogSession.boundedAlsoBy(giveUpAt);
                try {
                    writeOperation.run(txn);
                    committing=true;
@@ -1631,10 +2816,16 @@
                    failure=e;
                    throw e;
                } finally { // the comment connection lives no longer than the trees it stamped, and no longer
                    // than the attempt that opened it: a replay stamps on a session of its own
                    // than the attempt that opened it: a replay stamps on a session of its own. The catalog
                    // connection goes with it, having written every row this attempt had to enrol - and
                    // committed each of them, so a replay finds them recorded and writes none again
                    partlyCommitted=txn.partlyCommitted;
                    try {
                        txn.stampSession.close();
                        try {
                            txn.stampSession.close();
                        } finally {
                            txn.catalogSession.close();
                        }
                    } catch (RuntimeException e) {
                        //the stamp is a diagnostic aid and must not become the outcome of the write: an unchecked
                        //throw out of a driver's close() would otherwise replace the failure being unwound (JLS
@@ -2068,39 +3259,39 @@
            return isExistsTable(treeName);
        }

        /**
         * Where an unqualified name of this transaction's connection resolves, asked of it once. Every
         * lookup of a table below is narrowed to it - the reason is in {@link TableScope} - and asking
         * per lookup would cost a round trip per tree of the backend on every open: pgjdbc answers both
         * halves of it with a select of its own. A transaction holds one connection for the whole of its
         * life, so one answer serves it all.
         */
        // not private: a private member is not inherited, and the writeable transaction below asks
        // for the scope of its own lookups through it
        TableScope tableScope;

        TableScope takeTableScope() {
            // a connection that would not answer is asked again rather than latched: what it says
            // decides a create and a drop, and one refused question would otherwise leave every lookup
            // of this transaction as wide as the whole server. Only the first refusal of a transaction
            // is reported: the ones behind it are the same connection saying the same thing, once per
            // tree of the backend
            if (tableScope==null || !tableScope.answered) {
                tableScope=TableScope.of(JDBCStorage.this, con, tableScope==null);
            }
            return tableScope;
        }

        // Readable, not writeable: the caller that asks about a tree this backend does not own is
        // the compressed schema migration (#873), which probes the shared tree from the writeable
        // transaction of RootContainer.open() but must not create or enrol it. Answering that from
        // the readable transaction keeps the probe available to every reader, and costs nothing:
        // the writeable one inherits it.
        // The name it asks about is the non-enrolling one for that same reason, and the question is
        // narrowed to where an unqualified name of this connection resolves, like every other table
        // lookup of this class: see isExistsTable(Connection, TableScope, String).
        boolean isExistsTable(TreeName treeName) {
            final String tableName = readTableName(treeName);
            // the catalog lookup guarding a create table is bounded as the operation it is, not as
            // the bulk statement it guards, and not as the class of the transaction that happens to
            // ask: it reads a data dictionary rather than the data, so a wait here is the metadata
            // lock of another session
            try {
                return bounded(con, StatementBound.OPERATION, () -> {
                    final DatabaseMetaData metaData = con.getMetaData();
                    // asked of the catalog by name: openTree(createOnDemand) calls this for every tree
                    // of the backend - about 25 of them for a stock suffix, on every open - and listing
                    // every table of the database each time costs the whole catalog once per tree, on a
                    // database this backend may well be sharing with something else
                    try (final ResultSet rs = metaData.getTables(null, null,
                            storedIdentifier(metaData, tableName), new String[]{"TABLE"})) {
                        while (rs.next()) {
                            // the name still has to be compared: "_" is a single-character wildcard in a
                            // metadata pattern, so "opendj_<hash>" also matches a table named "opendjX<hash>"
                            if (tableName.equalsIgnoreCase(rs.getString("TABLE_NAME"))) {
                                return true;
                            }
                        }
                    }
                    return false;
                });
            } catch (Exception e) {
                throw new StorageRuntimeException(e);
            }
            return JDBCStorage.this.isExistsTable(con, takeTableScope(), readTableName(treeName));
        }
    }
    /**
@@ -2120,6 +3311,10 @@
        // write() (and by ImporterImpl.close()) when the transaction is done with.
        final StampSession stampSession=new StampSession();

        // The connection the catalog rows of this transaction are written on, opened at the first
        // row there is to write and closed with the transaction, like the stamp session above.
        final CatalogSession catalogSession=new CatalogSession();

        /**
         * Whether this transaction has committed part of its own work, which takes the attempt out of the
         * replay of {@link JDBCStorage#write}: what it did no longer rolls back as a whole, and a
@@ -2200,6 +3395,18 @@
        public void openTree(TreeName treeName, boolean createOnDemand) {
            if (createOnDemand) {
                checkReadOnly();
                // what makes this tree nameable by a process which has opened nothing: see
                // getCatalogTree(). Written before the table and not after it, on a connection of the
                // catalog's own and committed there, so that the table is never there without a row
                // naming it - on every engine, and not only on the ones whose DDL happens to carry the
                // row along - and so that none of it commits the work of this transaction. Of the two
                // ways a half-done open can end, a catalog naming a table that is not there is the one
                // the removal is ready for - it skips such a row and says so - while a table nothing
                // names is adopted with its stale rows by the next open of that tree and is dropped by no
                // clear ever after. deleteTree() takes the row out after the drop for that same reason,
                // which is why it is not the mirror of this. It writes, so it comes after the read-only
                // check and not before it (#874)
                enrolInCatalog(treeName);
                // Every statement below is a DDL that commits, and each raises partlyCommitted through
                // commitStatement() rather than once for the method: every one of them is guarded by a
                // catalog read, so on an existing backend this method issues nothing at all. Raising the
@@ -2240,7 +3447,7 @@
                }else if (driverName.contains("oracle")) {
                    try {
                        // oracle has no "create index if not exists"; unquoted identifiers are stored in uppercase
                        if (!isExistsIndex(tableName.toUpperCase(),"k_"+tableName.substring("opendj_".length()))) {
                        if (!isExistsIndex(tableName.toUpperCase(Locale.ROOT),"k_"+tableName.substring("opendj_".length()))) {
                            commitStatement("create index k_"+tableName.substring("opendj_".length())+" on "+tableName+" (k)", true);
                        }
                    }catch (SQLException e) {
@@ -2254,12 +3461,341 @@
            }
        }

        /**
         * Records the tree in the catalog of this backend, creating the catalog itself along the way
         * when this is the first tree of the storage. Enrolling is the business of
         * openTree(createOnDemand) alone: naming a tree in order to read it must never put it up for
         * removal, since the tree read may belong to another backend of the same database - the
         * unqualified compressed schema trees such a database may still hold, say (#873).
         * <p>
         * The row is written whenever the catalog does not already record this tree at this table -
         * and not only when the table is created - so that a backend of an installation upgraded to a
         * version keeping a catalog fills it in at its first read-write open instead of waiting for
         * its trees to be created again. What the catalog already records is read once, when this
         * storage first opens it; see {@link #enrolledTrees}.
         * <p>
         * The row is written on a connection of the catalog's own and committed there, never on the one
         * this transaction runs on. It has to be committed: the open which fills the catalog of a
         * backend upgraded from a version keeping none creates no table at all, so there is nothing
         * else of {@link #openTree} to carry those rows, and a transaction failing after them would
         * take every one back - leaving the tables named by nothing and the next clear dropping
         * nothing, which is #888 over again. And that commit must not be this transaction's:
         * {@code RootContainer.open()} opens every tree of every base DN in a single write, a commit
         * anywhere inside it takes the whole write out of the replay - {@link #replayReason} reads
         * {@link #partlyCommitted} before it asks anything else - and a deadlock at the twentieth tree
         * would then fail the backend open where master replayed it. A connection of its own is what
         * gives the row a commit that is not the caller's.
         */
        void enrolInCatalog(TreeName treeName) {
            final TreeName catalog=getCatalogTree();
            if (catalog.equals(treeName)) {
                return; // the catalog holds no row of its own: catalogTables() adds it when its table is there
            }
            if (SHARED_COMPRESSED_SCHEMA_BASE_DN.equals(treeName.getBaseDN())) {
                return; // a tree this backend may not be the only owner of: see the constant
            }
            openCatalog(catalog);
            if (enrolledTrees.contains(treeName)) {
                return; // already recorded, at the table this open would record it at
            }
            try {
                catalogSession.transaction().upsert(catalog,
                    ByteString.valueOfUtf8(treeName.toString()),
                    ByteString.valueOfUtf8(getTableName(treeName)));
                // committed where it is written, so that the row is there before the table on every
                // engine and not only where the "create table" below happens to carry it - and on the
                // catalog's own connection, so that this commit is none of the caller's: see above
                catalogSession.commit();
                enrolledTrees.add(treeName);
            } catch (SQLException | RuntimeException e) {
                // the unchecked one as well, exactly as unenrolFromCatalog() takes it: upsert() answers a
                // failed statement with a StorageRuntimeException of its own, and what that statement left
                // behind has to be rolled back all the same. This connection outlives the row that failed
                // on it and carries every remaining tree of this open - postgres refuses every further
                // statement of a transaction whose statement failed (25P02), so a reset skipped here fails
                // the twenty-odd enrolments behind it with a cause nowhere near the one that started it
                catalogSession.reset();
                throw e instanceof StorageRuntimeException ? (StorageRuntimeException) e : new StorageRuntimeException(e);
            }
        }

        /**
         * The connection the catalog is read and written on, established where this transaction has
         * not established it yet.
         * <p>
         * The unchecked failure of the connect is taken like the checked one, the way every other
         * catalog path of this class takes it: {@link JDBCStorage#newCatalogConnection} hands on a
         * driver's unchecked answer to a connect it will not make as the unchecked failure it is, so a
         * catch of {@code SQLException} alone would let that one past unwrapped and without the line
         * saying which connection of this backend could not be made.
         */
        Connection catalogConnection() {
            try {
                return catalogSession.connection();
            } catch (SQLException | RuntimeException e) {
                throw e instanceof StorageRuntimeException ? (StorageRuntimeException) e
                    : new StorageRuntimeException("jdbc: backend "+config.getBackendId()
                        +" could not open the connection its tree catalog is read and written on", e);
            }
        }

        /**
         * Makes the catalog of this backend usable, once per open of the storage: its table is created
         * where there is none, and what it already records is read where there is one.
         * <p>
         * Serialized on the storage, so that two transactions opening trees at the same time cannot both
         * find the table absent and both go on to create it. It serializes this storage and nothing
         * else, which is why the create tolerates a table that turned up while it was being made: an
         * offline tool beside a running server is a pair no lock of one process can order. The stamp -
         * the one thing under it that is nobody's dependency - is issued outside it.
         * <p>
         * Two things are kept out of the lock because they are the slow ones. The flag is read before
         * it is taken at all, which is every {@code openTree} of this storage but the first few: it is
         * volatile and it is raised after {@link #enrolledTrees} has been filled, so a reader that sees
         * it up sees that memo whole. And the connection of the catalog is established before it: that
         * connect retries a database taking no connection for the moment for up to the deadline of a
         * borrow (see {@link JDBCStorage#newCatalogConnection}), and made under the lock it would hold
         * every other transaction of this storage that goes on to open a tree for the whole of that
         * wait - a queue the borrows of the pool, each waiting on its own thread, never form.
         * <p>
         * What that costs is a connect to every transaction which finds the flag down and then loses the
         * race for the lock. The loser gives that connection up rather than hold it: the winner has
         * filled {@link #enrolledTrees}, so the caller is about to find its tree recorded and write
         * nothing at all, and the session is lazy - the rarer loser that does have a tree to enrol opens
         * another. The race is for the first openTree of a storage, so what this can cost against a
         * database with no connection to give is one refused login per racing transaction, where the
         * connect made under the lock cost one and made the others wait out the same refusal in turn.
         */
        void openCatalog(TreeName catalog) {
            if (catalogTableOpened) {
                return;
            }
            // what this call established, and not what the transaction was already holding: deleteTree()
            // opens the session before it drops anything, so a later openTree of the same transaction
            // must not give away a connection it did not make
            final boolean established=!catalogSession.isEstablished();
            catalogConnection();
            final boolean lostTheRace;
            synchronized (catalogLock) {
                lostTheRace=catalogTableOpened;
                if (!lostTheRace) {
                    if (isExistsTable(catalog)) {
                        readEnrolledTrees(catalog);
                    } else {
                        createCatalogTable(catalog);
                        // nothing to read from a table that has just been created, and nothing this open
                        // enrols may be skipped as already recorded
                    }
                    catalogTableOpened=true;
                }
            }
            if (lostTheRace) {
                if (established) {
                    // the winner has filled enrolledTrees, so the caller is about to find its tree recorded
                    // and write nothing: the connection this call made is given up rather than held idle for
                    // the rest of the transaction, and the session being lazy, the rarer loser that does
                    // have a tree to enrol opens another. Outside the lock, for the reason the connect is:
                    // a close is a round trip of its own, and against a database that has stopped answering
                    // it does not return at all - connectCatalog() lifts the read bound of the login on
                    // every connection it hands back, so there is no bound of ours left to end this one
                    catalogSession.close();
                }
                return;
            }
            // stamped with its tree name like any table of a tree (#866), and for a reason of its own: a
            // clear reports what it did not drop, and the catalog of a backend sharing this database
            // (#873) is the one table such a report could otherwise attribute to nobody. It costs one
            // stamp per open of the storage, not one per tree - the flag above is what keeps it to one -
            // and it is issued outside the lock: it is a diagnostic aid on a session and a bound of its
            // own, with no business holding up every openTree of this storage
            commentTable(catalog, dialectOf(con), stampSession);
        }

        /**
         * Reads what the catalog already records, so that the trees it names are not enrolled again on
         * an open which would write the rows that are already there; see {@link #enrolledTrees}. Run
         * once per open of the storage, behind the very flag that keeps the catalog from being opened
         * again, and it costs the one select a clear pays for anyway.
         * <p>
         * Read on the catalog's own connection and committed there, so that no transaction of a caller
         * ever touches the catalog table. A select of the caller's transaction would hold a lock on it
         * until that transaction ended - the whole of {@code RootContainer.open()} - and the rows this
         * read decides are written on the catalog's connection: a clear of this backend queueing for the
         * table in between would then be waiting for the caller while the caller waited for it, a pair
         * of sessions no deadlock detector of the database can see, one of them being blocked inside
         * this process rather than in the server. {@link #removeStorageFiles()} and {@link #listTrees()}
         * read that table on a connection of their own, which is the same argument read the other way:
         * neither is inside a transaction of a caller, and both are done with it when they commit.
         */
        void readEnrolledTrees(TreeName catalog) {
            try {
                final Connection catalogCon=catalogSession.connection();
                for (final Map.Entry<TreeName,String> row : readCatalogRows(catalogCon, getTableName(catalog)).entrySet()) {
                    // a row recording another table than this version would record is not the row this
                    // open would leave behind: a removal drops the table the row records, so such a row is
                    // rewritten - and committed - exactly like one that is not there at all. Asked through
                    // the non-enrolling name of #881: reading what the catalog records is not taking an
                    // interest in the tree it names, and a row this decides not to trust must not have put
                    // its tree in the memo of the trees this backend names its tables for
                    if (readTableName(row.getKey()).equals(row.getValue())) {
                        enrolledTrees.add(row.getKey());
                    }
                }
                catalogCon.commit(); // the read ends here and holds nothing of the catalog after it
            } catch (SQLException | RuntimeException e) {
                // the unchecked one as well, for the reason enrolInCatalog() takes it: this connection is
                // the one every enrolment of this open goes on to write its row on
                catalogSession.reset();
                throw e instanceof StorageRuntimeException ? (StorageRuntimeException) e : new StorageRuntimeException(e);
            }
        }

        /**
         * Creates the table of the catalog, on the catalog's own connection for the reason its rows are
         * written there - see {@link #enrolInCatalog}. It takes no index of the kind openTree() gives a
         * tree: the catalog is read whole and written by key, never iterated by key range, so the index
         * a cursor needs would serve nothing here. The stamp it does take is given by the caller, on
         * every open rather than on creation alone.
         * <p>
         * A read-write open of a JDBC backend needs the privilege to create this table, where a version
         * keeping no catalog issued no DDL at all on an installation whose tables were already there.
         * An account that may write its rows but not create a table is a configuration this can meet,
         * so the failure says which table it was and why the backend wanted it, rather than reaching
         * the operator as a bare SQL error inside ERR_OPEN_ENV_FAIL.
         */
        void createCatalogTable(TreeName catalog) {
            final String tableName=getTableName(catalog);
            try {
                final Connection catalogCon=catalogSession.connection();
                try (final PreparedStatement statement=catalogCon.prepareStatement("create table "+tableName+" ("+getTableDialect()+")")) {
                    // bulk like every other create table of this backend (#882): it is DDL nobody waits on,
                    // and the class of a client operation is not what a statement of this kind can be given
                    execute(statement, StatementBound.BULK);
                }
                catalogCon.commit();
            } catch (SQLException | RuntimeException e) {
                // the unchecked one as well, for the reason enrolInCatalog() takes it: what the statement
                // left behind has to be rolled back whatever class the failure arrived in, this connection
                // being the one the rows of this open are written on
                catalogSession.reset();
                // a table that turned up between the lookup and this statement is what was wanted, whoever
                // made it: the lock this runs under orders the transactions of one storage, and an offline
                // tool beside a running server - the pair #888 is about - is ordered by nothing at all.
                // The lookup is asked inside a catch and must not become the answer: it goes to the
                // database on the caller's connection, which is often the very thing that has just failed,
                // and it reports its own failure as a StorageRuntimeException - thrown from here it would
                // replace the create failure below with a bare metadata error saying nothing about the
                // catalog. So a lookup that will not answer is carried by the failure it could not settle.
                boolean alreadyThere;
                try {
                    alreadyThere=isExistsTable(catalog);
                } catch (RuntimeException lookup) {
                    e.addSuppressed(lookup);
                    alreadyThere=false;
                }
                if (alreadyThere) {
                    logger.debug(LocalizableMessage.raw("jdbc: table %s was created by another session while this one was creating it: %s",
                        tableName, stackTraceToSingleLineString(e)));
                    return;
                }
                throw new StorageRuntimeException("jdbc: backend "+config.getBackendId()+" could not create table "
                    +tableName+", which holds the catalog naming the trees it owns: a read-write open of a JDBC"
                    +" backend needs the privilege to create it, and a clear of one names nothing without it", e);
            }
        }

        /**
         * Takes the tree out of the catalog: a row is what puts a table up for removal, and this one is
         * gone. Written and committed on the catalog's own connection, like the enrolment - see {@link
         * #enrolInCatalog} - which is what keeps the caller's transaction from being able to roll it
         * back over a table that is already dropped.
         */
        void unenrolFromCatalog(TreeName treeName, boolean enrolled) {
            final TreeName catalog=getCatalogTree();
            if (catalog.equals(treeName)) {
                catalogTableOpened=false; // its own table is gone: the next enrolment creates it again
                enrolledTrees.clear(); // and records every tree anew, this one having recorded nothing
                return;
            }
            if (SHARED_COMPRESSED_SCHEMA_BASE_DN.equals(treeName.getBaseDN())) {
                // the symmetry of enrolInCatalog() and nothing more: no row of this pair was ever written,
                // so the delete would find none. What keeps the pair out of a clear is that a clear drops
                // what the catalog names and the catalog does not name them; see the constant
                return;
            }
            if (!enrolled) {
                // no row to delete - the catalog table is not there at all - and nothing to order this
                // against: a tree the catalog does not name is not one an enrolment may skip
                enrolledTrees.remove(treeName);
                return;
            }
            try {
                // deleteRow() and not delete(): the read-only check belongs to the caller of deleteTree,
                // which made it, and the transaction this row is written through is one of this class's own
                catalogSession.transaction().deleteRow(catalog, ByteString.valueOfUtf8(treeName.toString()));
                catalogSession.commit();
            } catch (SQLException | RuntimeException e) {
                // the unchecked one as well: deleteRow() answers a failed statement with a
                // StorageRuntimeException, and what that statement left behind has to be rolled back all
                // the same - this connection outlives the row that failed on it
                catalogSession.reset();
                throw e instanceof StorageRuntimeException ? (StorageRuntimeException) e : new StorageRuntimeException(e);
            } finally {
                // taken out of what this storage knows the catalog records after the delete and never
                // before it, so that the memo and the catalog never disagree in the direction that
                // makes an enrolment write a row a committed delete then takes back out. After the
                // attempt whatever became of it: a delete that failed leaves a row the next enrolment
                // has to write again rather than skip as already recorded, which costs an upsert of a
                // row that is already there and no more.
                //
                // What no ordering of these two lines can do is order this against an openTree of the
                // very same tree on another thread, and it is worth saying which fix was ruled out
                // rather than leaving it to be proposed again. Such an openTree landing between the
                // commit above and this line skips its enrolment - the memo still names the tree - and
                // goes on to create the table, leaving a table nothing names; the ordering before this
                // one reached the same end state by the other route, the enrolment writing a row this
                // delete then removed. A lock over the memo and the row closes neither, since the
                // table is created and dropped outside it either way: only a lock held across the DDL
                // of both would, and that one deadlocks. A transaction holding catalogLock and blocked
                // in the database on a "drop table" of a tree a second transaction of this storage is
                // still writing would be waiting for that transaction, while it waited for the lock at
                // its next openTree - a cycle the database cannot see, where today it is a plain wait
                // that ends when the second transaction does.
                //
                // So the catalog is consistent given that no two transactions open and delete the same
                // tree at once, and that is the layer above's to keep: a tree is opened read-write and
                // deleted from the configuration framework, which orders the changes of one entry, or
                // from EntryContainer.clear() with the backend disabled.
                enrolledTrees.remove(treeName);
            }
        }

        /** Whether a delete of this tree has a row of the catalog to take out; see {@link #unenrolFromCatalog}. */
        boolean isEnrolledTree(TreeName treeName) {
            if (getCatalogTree().equals(treeName) || SHARED_COMPRESSED_SCHEMA_BASE_DN.equals(treeName.getBaseDN())) {
                return false;
            }
            return catalogTableOpened || isExistsTable(getCatalogTree());
        }

        /**
         * Whether the table already carries the index of this name, asked where the table itself is
         * asked for - see {@link TableScope}. A table name carries no backend id and no database, so two
         * databases of one server hold identical table <em>and</em> index names, and Connector/J 8 binds
         * no schema predicate for a null catalog: a neighbouring database answering here would skip the
         * create index of this one for good, leaving every "where k>? order by k" batch of every cursor
         * a full scan behind it.
         */
        boolean isExistsIndex(String tableName, String indexName) throws SQLException {
            final TableScope scope=takeTableScope();
            // the index lookup takes the operation bound of #882 like every other catalog read of this
            // class: it asks a data dictionary rather than the data, so a wait here is the metadata lock
            // of another session - and it is narrowed to the scope every table lookup here is narrowed to
            return bounded(con, StatementBound.OPERATION, () -> {
                // approximate=true: with false the oracle driver runs ANALYZE on every call
                try (final ResultSet rs = con.getMetaData().getIndexInfo(null, null, tableName, false, true)) {
                try (final ResultSet rs = con.getMetaData().getIndexInfo(scope.catalog, null, tableName, false, true)) {
                    while (rs.next()) {
                        if (indexName.equalsIgnoreCase(rs.getString("INDEX_NAME"))) {
                        if (indexName.equalsIgnoreCase(rs.getString("INDEX_NAME")) && scope.covers(rs)) {
                            return true;
                        }
                    }
@@ -2280,6 +3816,34 @@
        @Override
        public void deleteTree(TreeName treeName) {
            checkReadOnly();
            // The row is taken out on the catalog's own connection rather than left to this transaction:
            // that transaction is the last thing the delete could still be rolled back by - write()
            // replays a class 40 conflict and rethrows everything else unreplayed - and the row would be
            // rolled back over a table that is already gone, with nothing ever to put it right: a deleted
            // tree is not opened again, so no enrolment and no unenrolment reaches it a second time. It
            // holds for the branch where there is no table to drop as much as for the one where the drop
            // commits of its own accord.
            // That connection is opened here, before anything is dropped: a connect this backend cannot
            // make costs nothing at this point, where one failing after the drop would leave exactly the
            // half-done state the sentence above is about.
            final boolean enrolled=isEnrolledTree(treeName);
            if (enrolled) {
                catalogConnection();
            }
            // The table dropped is the one the tree names, where a clear drops the one its row records.
            // The two are the same table by the time anything is deleted: a row recording another one is
            // not taken as an enrolment - readEnrolledTrees() keeps it out of enrolledTrees - so the
            // openTree that every delete of a tree comes after has rewritten it to this name.
            // A row is written before its table is created and taken out after its table is dropped,
            // never the other way round: of the two ways a half-done change can end, a catalog naming a
            // table that is not there is the one the removal is ready for - it skips such a row and says
            // so - while a table nothing names is adopted with its stale rows by the next open of that
            // tree and is dropped by no clear ever after. So this is deliberately not the mirror of
            // openTree(): an unenrolment left pending before the drop would be committed by the drop
            // itself on mysql and oracle, where DDL commits the transaction it finds open before it
            // executes, and would then stand even where the drop goes on to fail - ORA-00054 on a tree
            // another session holds, say, which write() does not replay, it being neither a class 40
            // state nor ORA-00060.
            if (isExistsTable(treeName)) {
                try {
                    commitStatement("drop table " + getTableName(treeName), true);
@@ -2287,7 +3851,8 @@
                    throw new StorageRuntimeException(e);
                }
            }
            // forget the mapping so listTrees() consumers (updateTableStatistics) skip the dropped table
            unenrolFromCatalog(treeName, enrolled);
            // the memoized table name of a tree nothing holds any more is of no use to anyone
            tree2table.invalidate(treeName);
            unstampableTrees.remove(treeName); // a table recreated later deserves a fresh stamp attempt
        }
@@ -2382,6 +3947,16 @@
        @Override
        public boolean delete(TreeName treeName, ByteSequence key) {
            checkReadOnly();
            return deleteRow(treeName, key);
        }

        /**
         * The statement of {@link #delete} without its read-only check, for the rows this class writes
         * on a transaction of its own making: the catalog of a backend is written through a transaction
         * over a connection of its own, whose access mode is read again as it is built, and the check
         * that matters was made by the caller of {@code openTree} or {@code deleteTree}.
         */
        boolean deleteRow(TreeName treeName, ByteSequence key) {
            try (final PreparedStatement statement=con.prepareStatement("delete from "+getTableName(treeName)+" where h="+hashParam(con)+" and k=?")){
                statement.setString(1,key2hash.get(ByteBuffer.wrap(key.toByteArray())));
                statement.setBytes(2,real2db(key.toByteArray()));
@@ -2523,8 +4098,11 @@
                throw new UnsupportedOperationException();
            }
            if (writeTableName==null) {
                // the enrolling name, unlike the read statements above: this writes to the tree, so
                // it is one this backend owns, and removeStorageFiles() has to know about it
                // the enrolling name, unlike the read statements above: this writes to the tree, so it is
                // one this backend owns and its table belongs in the memo of the storage. What a clear
                // drops is what the catalog of the backend names (#888), and openTree(name, true) is the
                // one thing that writes there - a tree written through a cursor is one the backend opened
                // to get the cursor, which is where its row comes from
                writeTableName=getTableName(treeName);
            }
            try (final PreparedStatement statement=con.prepareStatement("delete from "+writeTableName+" where h="+hashParam(con)+" and k=?")){
@@ -2637,9 +4215,185 @@
        }
    }
    
    /**
     * {@inheritDoc}
     * <p>
     * Answered from the catalog of the backend rather than from the trees this process happens to
     * have touched: {@link #removeStorageFiles()} runs before anything has touched one (#888).
     * <p>
     * What a tool has to be shown is not what a clear may drop: the shared compressed schema trees
     * are deliberately not enrolled - a backend must not offer a tree another one may own for removal
     * - and would go unnamed by {@code dbtest} for it, so they are added here when their tables are
     * there. {@link #catalogTables(Connection, TableScope)} is what the removal reads, and it
     * names them not.
     * <p>
     * The catalog itself is among the names, being a tree of this backend like any other: {@code
     * dbtest list-raw-dbs} counts it and {@code dump-raw-db} resolves its name, which is the one way
     * of seeing from outside the server what a clear of this backend would drop.
     */
    @Override
    public Set<TreeName> listTrees() {
        return tree2table.asMap().keySet();
        // validated, like the borrows of open() and removeStorageFiles(): since the catalog this reads
        // from, this borrow issues its statements far from itself and compensates a dropped connection
        // in no other way - a write is replayed and a read tells the pool, and this does neither, so a
        // connection dropped inside the alive window would surface out of a listing of tree names
        try (final Connection con=getValidatedConnection()) {
            return listTrees(con);
        } catch (StorageRuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw new StorageRuntimeException(e);
        }
    }

    Set<TreeName> listTrees(Connection con) throws SQLException {
        final TableScope scope=TableScope.of(this, con);
        final Set<TreeName> trees=new HashSet<>(catalogTables(con, scope).keySet());
        for (final TreeName treeName : SHARED_COMPRESSED_SCHEMA_TREES) {
            // asked of the database, not assumed: the pair belongs to no backend in particular, and
            // once #881 gives each backend a pair of its own an installation may hold neither table.
            // Narrowed to this database: a pair of the same name in another database of the server
            // would otherwise have this backend name two trees it does not hold
            if (isExistsTable(con, scope, readTableName(treeName))) {
                trees.add(treeName);
            }
        }
        return trees;
    }

    /**
     * The trees the catalog of this backend names, each with the table recorded as holding it, the
     * catalog itself among them. Empty when the catalog table is not there - a backend which has
     * never been opened read-write - which is what tells {@link #removeStorageFiles()} it has nothing
     * it may drop.
     * <p>
     * The table name is taken from the row rather than recomputed from the tree name, so that a
     * removal drops what was enrolled even if the naming of tables were ever to change.
     * <p>
     * The scope is the caller's rather than asked for here: it is not free of a round trip - pgjdbc
     * answers both halves of it with a select of its own - and a clear and a listTrees() both narrow a
     * lookup of their own by it, so they pass what they have instead of every reader asking twice over.
     */
    Map<TreeName,String> catalogTables(Connection con, TableScope scope) throws SQLException {
        return catalogTables(con, scope, new ArrayList<>()); // nobody to tell: the descriptions go nowhere
    }

    /**
     * The same, telling the caller what the read passed over: a clear accounts for every row of its
     * catalog, and a row it could not act on is one nothing else in its report would name - the table
     * such a row records is outside the namespace {@link #leftoverTables} scans. See {@link
     * #reportClearOutcome}.
     */
    Map<TreeName,String> catalogTables(Connection con, TableScope scope, List<String> skippedRows) throws SQLException {
        final TreeName catalogTree=getCatalogTree();
        final String catalogTable=getTableName(catalogTree);
        // narrowed to this database: a catalog of the same name in another database of the server
        // would send the select below at a table that is not here, failing the clear it answers
        if (!isExistsTable(con, scope, catalogTable)) {
            return Collections.emptyMap();
        }
        final Map<TreeName,String> trees=readCatalogRows(con, catalogTable, skippedRows);
        // The catalog names every tree of the backend but itself, and is put last on purpose: the
        // removal drops the trees in this order, and what names them has to outlive them. Dropping a
        // table is DDL, which mysql and oracle commit as they go, so a removal that fails halfway is
        // finished by the next attempt rather than leaving behind tables nothing names any more.
        trees.remove(catalogTree); // no row should name it; one that does must not hold back the order
        trees.put(catalogTree, catalogTable);
        return trees;
    }

    /**
     * The rows of the catalog table as they stand, tree by tree: the caller has already established
     * that the table is there - {@link #catalogTables} by a lookup of its own, an enrolment by having
     * just created it or found it - so this asks the database nothing but the select.
     * <p>
     * A row this backend cannot have written is skipped and reported rather than trusted. What a clear
     * drops is the table a row records, dropped by that name, so a row recording something outside the
     * namespace this backend names its tables in points at a table that is nobody's business of this
     * one's - and a row naming no tree at all, or naming one that is not a tree name, would otherwise
     * fail every clear from here on rather than the one thing it describes.
     * <p>
     * Every row passed over is described into {@code skippedRows}, the warn above being addressed to
     * whoever is reading the log at that moment and this to the account a clear gives of itself: such
     * a row is a tree the clear cannot see, so what the row records is dropped by nothing - while the
     * row itself goes with the catalog table it sits in, which the clear names last and drops. That is
     * what makes the line the only surviving copy of what such a row said, and why it carries the
     * recorded name. A reader with nobody to tell - a read of {@code dbtest}, or the one an enrolment makes
     * - hands in a list of its own and lets it go, which is one allocation per read of a whole table
     * and no convention to get wrong.
     */
    Map<TreeName,String> readCatalogRows(Connection con, String catalogTable) throws SQLException {
        return readCatalogRows(con, catalogTable, new ArrayList<>());
    }

    Map<TreeName,String> readCatalogRows(Connection con, String catalogTable, List<String> skippedRows)
            throws SQLException {
        final Map<TreeName,String> trees=new LinkedHashMap<>();
        // the rows are read inside the bound rather than from a live ResultSet: #882 took the
        // executeResultSet() that returned one away, so a transfer cannot run with nothing bounding it
        try (final PreparedStatement statement=con.prepareStatement("select k,v from "+catalogTable)) {
            executeResultSet(statement, rs -> {
                while (rs.next()) {
                    final byte[] key=rs.getBytes("k");
                    if (key==null) { // no tree is named by a row with no key, and a clear must not fail over one
                        logger.warn(LocalizableMessage.raw("jdbc: table %s holds a row naming no tree at all: skipped",
                            catalogTable));
                        skippedRows.add("a row naming no tree at all");
                        continue;
                    }
                    final String name=new String(db2real(key), StandardCharsets.UTF_8);
                    final TreeName treeName;
                    try {
                        treeName=TreeName.valueOf(name);
                    } catch (RuntimeException e) { // reported rather than passed off as a backend with fewer trees
                        logger.warn(LocalizableMessage.raw("jdbc: table %s holds \"%s\", which is not the name of a tree: skipped",
                            catalogTable, name));
                        skippedRows.add("\""+name+"\", which is not the name of a tree");
                        continue;
                    }
                    final byte[] table=rs.getBytes("v");
                    final String tableName=table==null || table.length==0
                        ? readTableName(treeName) // a row of a version which recorded the name and not the table
                        : new String(table, StandardCharsets.UTF_8);
                    // The prefix and not the whole of the name: the table recorded is taken from the row
                    // rather than derived again so that a removal drops what was enrolled even if the naming
                    // of tables were ever to change, and every naming this backend could take up is inside
                    // the namespace it already scans for what a clear left standing. What the shape does have
                    // to rule out is anything that is not a bare identifier: this value is read back from a
                    // table and reaches a "drop table" that no driver will take a bind parameter for.
                    if (!isOwnTableName(tableName)) {
                        logger.warn(LocalizableMessage.raw("jdbc: table %s records tree %s at \"%s\", which is no table of this backend: skipped",
                            catalogTable, treeName, tableName));
                        skippedRows.add(treeName+" at \""+tableName+"\", which is no table of this backend");
                        continue;
                    }
                    trees.put(treeName, tableName);
                }
                return null;
            });
        }
        return trees;
    }

    /**
     * Whether a name read back from the catalog is one of this backend's tables: inside the namespace
     * it names them in, and a bare identifier besides. A clear drops the table a row records, by that
     * name, in a statement built by concatenation - the DDL of no engine here takes a bind parameter
     * for it - so a row is trusted to name a table of this backend and nothing else. The existence
     * lookup in front of the drop would answer no for most of what this rules out; it is not what
     * makes it safe.
     */
    static boolean isOwnTableName(String tableName) {
        if (!tableName.toLowerCase(Locale.ROOT).startsWith("opendj")) {
            return false;
        }
        for (int i=0;i<tableName.length();i++) {
            final char c=tableName.charAt(i);
            if (!(c>='a' && c<='z') && !(c>='A' && c<='Z') && !(c>='0' && c<='9') && c!='_' && c!='$') {
                return false;
            }
        }
        return true;
    }

    final class ImporterImpl implements Importer {
@@ -2739,10 +4493,12 @@
        }

        /**
         * Hands the connection back to the pool and closes the stamp session, whatever went before.
         * Returns the failure the caller is to report: the return rolls back, and the rollback
         * fails on exactly the connection whose commit just did, so the commit stays the exception
         * the caller sees and this one rides along with it instead of replacing it.
         * Hands the connection back to the pool and closes the sessions the transaction opened
         * beside it - the stamp one and the catalog one (#888), both outside the pool and neither
         * outliving the import that opened it - whatever went before. Returns the failure the
         * caller is to report: the return rolls back, and the rollback fails on exactly the
         * connection whose commit just did, so the commit stays the exception the caller sees and
         * this one rides along with it instead of replacing it.
         */
        private SQLException releaseConnection(SQLException failure) {
            try {
@@ -2761,7 +4517,11 @@
                    failure.addSuppressed(reported);
                }
            } finally {
                txw.stampSession.close();
                try {
                    txw.stampSession.close();
                } finally {
                    txw.catalogSession.close();
                }
            }
            return failure;
        }

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java

@@ -1668,6 +1668,28 @@
        assertEquals(CachedConnection.attemptSeconds(30, Long.MAX_VALUE), 30);
    }

    /**
     * A deadline so far off that naming it overflows is a wait with no end, and not one already
     * behind us: a borrow configured to wait practically forever would otherwise give up on its
     * first retryable failure, which is the opposite of what was asked for. The sum is what has to
     * be guarded and not only the product - a value under the clamp of the product can still name a
     * moment past the end of the epoch.
     */
    @Test
    public void testADeadlineTooFarOffToNameIsAWaitWithNoEnd() throws Exception {
        final long startedAt = System.currentTimeMillis();
        // 0 is the operator asking for no deadline at all
        assertEquals(CachedConnection.deadlineOf(startedAt, 0), Long.MAX_VALUE);
        // ... and so is a value whose milliseconds would not fit a long at all
        assertEquals(CachedConnection.deadlineOf(startedAt, Long.MAX_VALUE / 1000), Long.MAX_VALUE);
        // the one the product guard lets through, which is the largest value it does: a second under
        // the clamp, so the milliseconds of it still fit a long - by 1807 of them - while the moment
        // they name, counted from now, does not. Guarded by the sum alone
        assertEquals(CachedConnection.deadlineOf(startedAt, Long.MAX_VALUE / 1000 - 1), Long.MAX_VALUE);
        // and an ordinary value still names the moment it says
        assertEquals(CachedConnection.deadlineOf(startedAt, 60), startedAt + 60_000);
    }

    /** The connection string holds the credentials of the backend: a stall report must not carry them. */
    @Test
    public void testLoggedConnectionStringCarriesNoCredentials() throws Exception {
@@ -1768,6 +1790,18 @@
        assertFalse(stall.contains("S3cret"), stall);
        assertTrue(stall.contains("jdbc:postgresql://h:5432/db"), stall);
        assertTrue(stall.contains("4000 ms") && stall.contains("(3 attempts)"), stall);

        // and so is the stall of a connect made outside the pool - the connection the tree catalog of a
        // backend is written on (#888) - which is under the same rule and describes the same url
        final String outside = CachedConnection.outsidePoolStallMessage(url, "tree catalog", 3, 4000,
            new SQLException("FATAL: too many connections for " + url));
        assertFalse(outside.contains("S3cret"), outside);
        assertTrue(outside.contains("jdbc:postgresql://h:5432/db"), outside);
        assertTrue(outside.contains("4000 ms") && outside.contains("(3 attempts)"), outside);
        assertTrue(outside.contains("tree catalog"), outside);
        // and says what it is: a borrow of the pool is what this connect is not, and an operator
        // reading it must not be sent to the pool for a stall the pool has no part in
        assertFalse(outside.contains("pooled one"), outside);
    }

    /**

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CatalogConnectionTestCase.java

New file
@@ -0,0 +1,658 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.backends.jdbc;

import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.DirectoryServerTestCase;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import java.sql.Connection;
import java.sql.Driver;
import java.sql.DriverManager;
import java.sql.DriverPropertyInfo;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.sql.SQLTimeoutException;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;

import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;

/**
 * The connection the tree catalog of a backend is read and written on (#888): what it hands its
 * driver, and what it does with the connection it gets back. It is established outside the pool -
 * the caller of {@code openTree()} is holding a pooled connection already - so nothing the pool is
 * asserted on covers it, and a container suite covers it only where a container starts: it skips
 * itself whole otherwise, which leaves a bound nothing exercises.
 * <p>
 * No database is needed for any of it. The url is a postgresql one the pgjdbc driver cannot parse,
 * so {@code DriverManager} falls through to the probe of this class, while {@code
 * CachedConnection.ConnectDialect} still reads it as postgres - which is what makes the connect fill
 * in bounds at all, and what a url of an engine of nobody's would not.
 */
@SuppressWarnings("javadoc")
public class CatalogConnectionTestCase extends DirectoryServerTestCase {

    /**
     * What a caller with no replay above it hands the connect: the importer is the one such caller in
     * the product, and every case here that is not about the window itself asks the way it asks, so
     * that what it pins is the property and not a window of the test's own.
     */
    private static final long NO_REPLAY_WINDOW = Long.MAX_VALUE;

    private ProbeDriver probeDriver;

    @BeforeClass
    public void registerProbeDriver() throws SQLException {
        probeDriver = new ProbeDriver();
        DriverManager.registerDriver(probeDriver);
    }

    @AfterClass(alwaysRun = true)
    public void deregisterProbeDriver() throws SQLException {
        if (probeDriver != null) {
            DriverManager.deregisterDriver(probeDriver);
        }
    }

    /**
     * Nothing of one case reaches the next: the probe is a field of the class and a case that fails
     * before its finally would otherwise leave its stubbed connection, its refusals or the properties
     * of its last attempt to be read by whatever runs after it.
     */
    @BeforeMethod
    public void resetProbe() {
        probeDriver.lastProperties = null;
        probeDriver.answer = null;
        probeDriver.refusal = null;
        probeDriver.refusalsLeft.set(0);
        probeDriver.attempts.set(0);
        probeDriver.refusalDelayMs = 0;
        probeDriver.interruptOnAttempt = false;
    }

    private static JDBCStorage storageFor(String url) {
        final JDBCBackendCfg cfg = mockCfg(JDBCBackendCfg.class);
        when(cfg.getBackendId()).thenReturn("catalogProbe");
        when(cfg.getDBDirectory()).thenReturn(url);
        return new JDBCStorage(cfg, null);
    }

    /**
     * The bound of the connect is the one the operator configured for this backend's connects, and
     * not a literal of the code: a deployment whose login legitimately takes longer than the default
     * raises {@link CachedConnection#CONNECT_TIMEOUT_PROPERTY} for it, and a catalog connect bounded
     * tighter than that fails in 08001 - which is no conflict a write replays, so the backend stops
     * opening on an installation that opened before this connection existed.
     * <p>
     * Asked with no deadline over it, so that what the case pins is the configured bound alone: what
     * a deadline does to it is the case below.
     */
    @Test
    public void testTheCatalogConnectTakesTheConfiguredBound() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, "120");
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        try {
            probeDriver.lastProperties = null;
            storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW).close();
            assertBoundedAt(probeDriver.lastProperties, 120);
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * One attempt is never left to run past the deadline the retry of this connect is given, which is
     * the deadline of a borrow ({@link CachedConnection#POOL_TIMEOUT_PROPERTY}): an attempt bounded
     * looser than what is left of it would overrun it by a whole connect timeout. The pool bounds its
     * own attempts by exactly this rule, and a connect established the way a pooled one is takes it.
     */
    @Test
    public void testTheCatalogConnectIsNeverBoundedPastTheDeadlineOfItsRetry() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, "120");
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "20");
        try {
            storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW).close();
            // a range and not the 20 exactly: the bound is what is left of the deadline when the attempt
            // is made, so a pause of a second anywhere before it - a collection, the first touch of a
            // class on a loaded box - makes it 19, and the case is about the deadline and not the clock
            assertBoundedWithin(probeDriver.lastProperties, 15, 20);
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * A database taking no connection <em>for the moment</em> - at its connection limit, or still
     * recovering - is waited out rather than reported: one attempt loses a race the borrow beside it
     * wins, and this connect is on the critical path of the first read-write open of every backend.
     */
    @Test
    public void testTheCatalogConnectWaitsOutADatabaseTakingNoConnectionForTheMoment() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        // both, since both are read on every connect: an ambient connect bound would change the
        // per-attempt bound these cases run under without changing anything they assert on
        System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "60");
        probeDriver.refusal = new SQLException("too many clients already", "53300");
        probeDriver.refusalsLeft.set(2);
        try {
            storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW).close();
            assertEquals(probeDriver.attempts.get(), 3,
                "a connect refused for the moment was not retried the way a borrow of the pool retries it");
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * And everything else is the caller's to see rather than waited out behind its back: a password
     * that is not accepted does not become a minute of silence and then the same failure.
     * <p>
     * A guard rather than a regression test, and worth saying so: the head before this connect had a
     * retry made one attempt and reported it, so it satisfies this case by having no loop at all.
     * What the case is here for is the loop that does exist staying this narrow - a predicate widened
     * to any refusal turns a wrong password into a minute of silence per open, and nothing else in
     * this suite would notice.
     */
    @Test
    public void testTheCatalogConnectDoesNotRetryAFailureThatWillNotClear() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        // both, since both are read on every connect: an ambient connect bound would change the
        // per-attempt bound these cases run under without changing anything they assert on
        System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "60");
        probeDriver.refusal = new SQLException("password authentication failed", "28P01");
        probeDriver.refusalsLeft.set(Integer.MAX_VALUE);
        try {
            storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);
            fail("a connect that will not clear was retried instead of being reported");
        } catch (SQLException expected) {
            assertEquals(expected.getSQLState(), "28P01", "the failure of the driver was not the one reported");
            assertEquals(probeDriver.attempts.get(), 1, "a failure that will not clear was attempted more than once");
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * The wait ends at the deadline of a borrow, as a timeout by type and carrying the state of the
     * driver's own last refusal: a state of this code's making would be read by {@code write()} as a
     * connection the database dropped, and the retry must change no classification.
     */
    @Test
    public void testTheCatalogConnectGivesUpAtTheDeadlineOfABorrow() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        // both, since both are read on every connect: an ambient connect bound would change the
        // per-attempt bound these cases run under without changing anything they assert on
        System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1");
        // a vendor code beside the state, since both are carried over: an oracle failure says what it
        // is in the ORA number and not in the SQLState, so a timeout dropping the code would answer 0
        // where the classifier reading it expects the driver's own
        probeDriver.refusal = new SQLException("the database system is starting up", "57P03", 3113);
        probeDriver.refusalsLeft.set(Integer.MAX_VALUE);
        try {
            storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);
            fail("a connect refused for the whole deadline was not given up on");
        } catch (SQLTimeoutException expected) {
            // the state of the driver's own refusal and not one of this code's making: a manufactured
            // 08001 is read by write() as a connection the database dropped, which would replay an
            // attempt whose pooled connection is healthy and distrust the pool over it
            assertEquals(expected.getSQLState(), "57P03",
                "the failure the deadline ended carried another state than the driver's own");
            assertEquals(expected.getErrorCode(), 3113,
                "the failure the deadline ended dropped the vendor code of the driver's own");
            assertNotNull(expected.getCause(), "the driver's own failure was not carried as the cause");
            assertTrue(probeDriver.attempts.get() > 1,
                "the deadline was reached without the connect having been retried at all");
            // and it says which of the two bounds ran out, the property being the thing to raise only
            // where the property is what ended the wait
            assertTrue(expected.getMessage().contains(CachedConnection.POOL_TIMEOUT_PROPERTY + "=1s"),
                "the timeout did not name the bound that ended it: " + expected.getMessage());
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * The wait may not outlast the replay window of the {@code write()} it runs inside: this loop
     * runs within one attempt of that one, so a refusal waited out past the window reaches it with
     * the window already spent and is thrown unreplayed - the retry would cost the caller the replay
     * it had before there was a retry here at all. The shorter of the two bounds is the deadline of
     * the wait; the bound of one attempt is not taken from it, and this case pins that too.
     */
    @Test
    public void testTheCatalogConnectDoesNotOutlastTheReplayWindowOfItsCaller() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        // a minute, which is the default and six times the window of a write: the property is what
        // this connect would wait out if the window did not reach it
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "60");
        probeDriver.refusal = new SQLException("the database system is starting up", "57P03");
        probeDriver.refusalsLeft.set(Integer.MAX_VALUE);
        final long startedAt = System.currentTimeMillis();
        try {
            storageFor(ProbeDriver.URL).newCatalogConnection(startedAt + 300);
            fail("a connect refused past the replay window of its caller was not given up on");
        } catch (SQLTimeoutException expected) {
            final long waited = System.currentTimeMillis() - startedAt;
            assertTrue(waited < 30_000,
                "the connect waited " + waited + " ms, which is the pool timeout rather than the window above it");
            assertTrue(probeDriver.attempts.get() > 1,
                "the window was spent without the connect having been retried at all");
            assertEquals(expected.getSQLState(), "57P03",
                "the failure the window ended carried another state than the driver's own");
            // the line has to send an operator to the right knob: raising the pool timeout moves
            // nothing where the window of the write is the shorter bound
            assertTrue(expected.getMessage().contains("replay window"),
                "the timeout did not say which of the two bounds ended it: " + expected.getMessage());
            // and one attempt keeps the bound the operator configured for a login of this database:
            // the window decides how long it is worth retrying, not how long a login may take, and
            // an attempt cut to what is left of the window is the connect dying where the pooled one
            // beside it succeeds - the backend that stops opening
            assertBoundedAt(probeDriver.lastProperties, CachedConnection.DEFAULT_CONNECT_TIMEOUT_SECONDS);
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * A thread asked to stop is not answered by sleeping out the rest of a pool timeout: the flag is
     * put back - every frame above this one reads it to decide whether to unwind - and the driver's
     * own refusal is what the caller is told, the interrupt riding along with it so that a connect
     * cut short by a shutdown is not read off the log as a database refusing connections.
     */
    @Test
    public void testTheCatalogConnectReportsAnInterruptRatherThanSleepingPastIt() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "60");
        probeDriver.refusal = new SQLException("the database system is starting up", "57P03");
        probeDriver.refusalsLeft.set(Integer.MAX_VALUE);
        // raised inside the attempt rather than by another thread racing this one: the first backoff
        // is a millisecond, and a sleep entered with the flag already up throws at once
        probeDriver.interruptOnAttempt = true;
        try {
            storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);
            fail("a connect interrupted while it waited was not reported at all");
        } catch (SQLException expected) {
            assertFalse(expected instanceof SQLTimeoutException,
                "an interrupt was reported as the deadline of the wait running out");
            assertEquals(expected.getSQLState(), "57P03",
                "the interrupt replaced the driver's own failure instead of riding along with it");
            assertEquals(probeDriver.attempts.get(), 1, "the wait went on past the interrupt");
            boolean carried = false;
            for (final Throwable suppressed : expected.getSuppressed()) {
                carried |= suppressed instanceof InterruptedException;
            }
            assertTrue(carried, "the interrupt was dropped rather than carried on the failure reported");
            assertTrue(Thread.currentThread().isInterrupted(),
                "the flag Thread.sleep() cleared was not put back, so nothing above can read it");
        } finally {
            Thread.interrupted(); // cleared here, or every case running after this one on this thread meets it
            probeDriver.interruptOnAttempt = false;
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * A connect of this backend that stalls says so in the log, and is throttled apart from the
     * borrows of the pool that address the same database: the two stall for the same reason, and a
     * borrow that reported a moment ago would otherwise silence the connect that is about to fail -
     * which is the one of the two an operator has no other line about.
     * <p>
     * The stall is a real one rather than a call of the formatter: the guard of the throttle passes
     * only where a connect has been retrying for a second, so nothing under it is reached by a case
     * whose refusals come back at once.
     */
    @Test
    public void testAStallOfTheCatalogConnectIsThrottledApartFromTheBorrowsOfThePool() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "60");
        // one refusal, slow enough that the connect has been retrying for longer than the guard of
        // the throttle, and then a connection: what is asserted is the warning, not the failure
        probeDriver.refusal = new SQLException("the database system is starting up", "57P03");
        probeDriver.refusalsLeft.set(1);
        probeDriver.refusalDelayMs = CachedConnection.STALL_WARNING_AFTER_MS + 100;
        try {
            // a url of its own: the throttle is keyed by url, and a case sharing one with another
            // would read that one's stamp instead of its own
            storageFor(ProbeDriver.STALL_URL).newCatalogConnection(NO_REPLAY_WINDOW).close();
            final long now = System.currentTimeMillis();
            final long longEnoughAgo = now - 2 * CachedConnection.STALL_WARNING_AFTER_MS;
            // the connect reported its stall: the moment is filed, so the next one inside the interval
            // is not due. Nothing else of this suite touches this url
            assertFalse(CachedConnection.stallWarningDue(ProbeDriver.STALL_URL, "|tree catalog", longEnoughAgo, now),
                "the connect stalled for longer than the guard and reported nothing");
            // and a borrow of the pool on that very url is still due one of its own, which is the half
            // of the throttle key that keeps the two waits from silencing each other
            assertTrue(CachedConnection.stallWarningDue(ProbeDriver.STALL_URL, "", longEnoughAgo, now),
                "a stall of this connect silenced the borrows of the pool addressing the same database");
        } finally {
            probeDriver.refusalDelayMs = 0;
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * Nothing configured is the default of the pool, which is what this connect used to take always.
     * The deadline is pinned rather than left to its own default: the attempt takes the shorter of
     * the two, so a pool timeout set anywhere - the surefire configuration, the environment, another
     * suite - would otherwise decide what this case asserts.
     */
    @Test
    public void testTheCatalogConnectTakesTheDefaultWhereNothingIsConfigured() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        try {
            probeDriver.lastProperties = null;
            storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW).close();
            assertBoundedAt(probeDriver.lastProperties, CachedConnection.DEFAULT_CONNECT_TIMEOUT_SECONDS);
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * A property of 0 is the operator asking for no bound at all - the pool reads it that way - and a
     * connect that bounded itself anyway would be answering a setting with the opposite of it. Nothing
     * is handed to the driver then, and there is no read bound to lift once the login is through.
     * <p>
     * Both properties, which is what the pool itself says leaves a connect unbounded: the deadline of
     * the retry bounds the attempt where there is one, so turning the per-attempt bound off alone
     * leaves the attempt bounded by what is left of that deadline - the case above.
     */
    @Test
    public void testTheCatalogConnectIsUnboundedWhereTheOperatorTurnedTheBoundOff() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, "0");
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        try {
            probeDriver.lastProperties = null;
            final Connection con = storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);
            con.close();
            assertNotNull(probeDriver.lastProperties, "no properties were handed to the driver at all");
            assertTrue(probeDriver.lastProperties.isEmpty(),
                "a connect the operator asked for no bound on was bounded anyway: " + probeDriver.lastProperties);
            verify(con, never()).setNetworkTimeout(any(), anyInt());
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * What the connection is handed back for: rows of its own, committed where they are written. The
     * read bound of the login is lifted as soon as the login is through (#872) - it is a bound of the
     * connect and not of the statements of the catalog - and the isolation is the pool's, a repeatable
     * read gap-locking a catalog two transactions enrol into.
     */
    @Test
    public void testTheCatalogConnectionIsSetUpForItsRows() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        // the read bound is lifted only where the attempt was given one, and the attempt takes the
        // shorter of the two properties: a deadline of 0 elsewhere would leave nothing to lift here
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        try {
            final Connection con = storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);
            con.close();
            verify(con).setAutoCommit(false);
            verify(con).setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
            verify(con).setNetworkTimeout(any(), eq(0));
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * A driver that will not take the read bound of the login back does not cost the backend its
     * catalog: the pool meets the same failure and hands the connection to the borrower waiting for
     * it, and a connect failing where the pooled one beside it succeeds is a backend that stops
     * opening on an installation which opened before this connection existed. Reported, and kept.
     * <p>
     * A guard rather than a regression test, like the one above: the head before this round already
     * caught that failure and returned the connection, so nothing of the round makes this case pass.
     * It is here because the answer was reached for twice - once as "fail the connect", which this
     * round took back out - and a third attempt at it would go unnoticed otherwise.
     */
    @Test
    public void testTheCatalogConnectionIsKeptWhereTheReadBoundWillNotComeOff() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        final Connection keeping = mock(Connection.class);
        doThrow(new SQLFeatureNotSupportedException("no network timeout here"))
            .when(keeping).setNetworkTimeout(any(), anyInt());
        probeDriver.answer = keeping;
        try {
            final Connection con = storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);
            assertNotNull(con, "a connection whose read bound would not come off was not handed back");
            verify(con).setAutoCommit(false);
            verify(con, never()).close();
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * A connection whose set-up failed is held by nobody - the caller is answered with the failure -
     * so it is closed here or it leaks for the life of the process, one per open of a storage.
     */
    @Test
    public void testAConnectionWhoseSetUpFailsIsClosed() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        // pinned like every other case of this class: both are read from the system properties on every
        // connect, so an ambient value would have this case exercise another path than the one it names
        System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        final Connection failing = mock(Connection.class);
        doThrow(new SQLException("no transaction here", "08006")).when(failing).setAutoCommit(false);
        probeDriver.answer = failing;
        try {
            storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);
            fail("a connection this backend could not set up was handed to the catalog");
        } catch (SQLException expected) {
            // the failure of the set-up itself, reported to the caller rather than swallowed
        } finally {
            probeDriver.answer = null;
            restore(previous);
            restorePool(previousPool);
        }
        verify(failing).close();
    }

    /** What the dialect of this url declares, at the given number of seconds. */
    private static void assertBoundedAt(Properties handed, long seconds) {
        assertNotNull(handed, "no properties were handed to the driver at all");
        final CachedConnection.ConnectDialect dialect = CachedConnection.ConnectDialect.of(ProbeDriver.URL);
        assertNotNull(dialect, "the url of this test is read as an engine of nobody's, so it is bounded by nothing");
        for (final String property : dialect.connectProperties) {
            assertEquals(handed.getProperty(property), Long.toString(seconds * dialect.connectUnitsPerSecond),
                property + " did not reach the driver at the configured bound");
        }
        assertEquals(handed.getProperty(dialect.readProperties[0]),
            Long.toString(seconds * dialect.readUnitsPerSecond),
            dialect.readProperties[0] + " did not reach the driver at the configured bound");
    }

    /** The same where the value is what is left of a deadline, which no case may pin to the millisecond. */
    private static void assertBoundedWithin(Properties handed, long atLeastSeconds, long atMostSeconds) {
        assertNotNull(handed, "no properties were handed to the driver at all");
        final CachedConnection.ConnectDialect dialect = CachedConnection.ConnectDialect.of(ProbeDriver.URL);
        assertNotNull(dialect, "the url of this test is read as an engine of nobody's, so it is bounded by nothing");
        final String property = dialect.connectProperties[0];
        final String handedValue = handed.getProperty(property);
        assertNotNull(handedValue, property + " did not reach the driver at all");
        final long seconds = Long.parseLong(handedValue) / dialect.connectUnitsPerSecond;
        assertTrue(seconds >= atLeastSeconds && seconds <= atMostSeconds,
            property + " reached the driver at " + seconds + "s, outside the deadline it is taken from ("
                + atLeastSeconds + ".." + atMostSeconds + "s)");
    }

    private static void restore(String previous) {
        if (previous == null) {
            System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        } else {
            System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, previous);
        }
    }

    private static void restorePool(String previous) {
        if (previous == null) {
            System.clearProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        } else {
            System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, previous);
        }
    }

    /** Records the properties a catalog connection hands its driver, and connects to nothing. */
    private static final class ProbeDriver implements Driver {
        /**
         * A postgresql url with a port that is not a number: pgjdbc cannot parse it and answers the
         * DriverManager with null - or with a failure, which it records and walks past all the same -
         * so this probe is the driver that ends up answering, while ConnectDialect still reads the
         * prefix as postgres.
         */
        static final String URL = "jdbc:postgresql://catalog-probe:not-a-port/db";

        /**
         * The same for the one case about the stall warning, which reads the throttle this connect
         * files its report in: that throttle is keyed by url, so a case sharing one with any other
         * would be asserting on whichever of them ran first.
         */
        static final String STALL_URL = "jdbc:postgresql://catalog-probe-stall:not-a-port/db";

        volatile Properties lastProperties;

        /** The connection to answer with, for a test about what is done with it; a fresh mock otherwise. */
        volatile Connection answer;

        /** How many attempts to refuse before answering, and with what; for the cases about the retry. */
        final AtomicInteger refusalsLeft = new AtomicInteger();
        volatile SQLException refusal;

        /** How long a refused attempt takes, for the one case that needs a wait the throttle counts. */
        volatile long refusalDelayMs;

        /**
         * Whether an attempt raises the interrupt flag of the thread asking for it, so that the case
         * about an interrupted wait does not have to race a backoff of one millisecond from outside.
         */
        volatile boolean interruptOnAttempt;

        /** Every attempt this driver was asked to make, refused ones included. */
        final AtomicInteger attempts = new AtomicInteger();

        @Override
        public Connection connect(String url, Properties info) throws SQLException {
            if (!acceptsURL(url)) {
                return null; // not ours: DriverManager goes on to the next driver
            }
            attempts.incrementAndGet();
            lastProperties = info;
            if (refusal != null && refusalsLeft.getAndDecrement() > 0) {
                if (refusalDelayMs > 0) {
                    try {
                        Thread.sleep(refusalDelayMs);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
                if (interruptOnAttempt) {
                    Thread.currentThread().interrupt();
                }
                throw refusal;
            }
            return answer != null ? answer : mock(Connection.class);
        }

        @Override
        public boolean acceptsURL(String url) {
            return url != null && (url.startsWith(URL) || url.startsWith(STALL_URL));
        }

        @Override
        public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) {
            return new DriverPropertyInfo[0];
        }

        @Override
        public int getMajorVersion() {
            return 1;
        }

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

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

        @Override
        public java.util.logging.Logger getParentLogger() throws SQLFeatureNotSupportedException {
            throw new SQLFeatureNotSupportedException();
        }
    }
}

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStorageRetryTest.java

@@ -41,6 +41,8 @@
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.logging.Logger;

import static org.forgerock.i18n.LocalizableMessage.raw;
@@ -49,6 +51,7 @@
import static org.mockito.Mockito.anyBoolean;
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -97,6 +100,13 @@
  private Connection engineConnection;

  /**
   * The connection the tree catalog of a storage of this test is written on, so that a test can assert what was
   * written there. Nothing else can: it is opened straight through the driver rather than borrowed from the pool,
   * and the rows it carries are the ones {@code statements} is asserted never to have carried.
   */
  private Connection catalogConnection;

  /**
   * Connections whose class names carry the engine the way the drivers' own do - pgjdbc's
   * {@code org.postgresql.jdbc.PgConnection}, Connector/J's {@code com.mysql.cj.jdbc.ConnectionImpl}. That name
   * is what {@code driverNameOf()} matches an engine on, and the name of a mock is derived from the type it
@@ -444,6 +454,45 @@
  }

  /**
   * The first read-write open of a backend upgraded from a version keeping no catalog creates the catalog and
   * writes one row per tree, and creates no table of its own: every tree it opens is already there. None of that
   * may commit anything of the caller's - {@code RootContainer.open()} opens every tree of every base DN in a
   * single write, and a commit anywhere inside it takes the whole open out of the replay for the life of that
   * attempt, so a deadlock at the twentieth tree would fail the backend start-up that master replayed. The rows
   * still have to be committed, since nothing else of this open would carry them: a connection of the catalog's
   * own is what makes the two compatible.
   */
  @Test
  public void testFillingTheCatalogOfAnUpgradedBackendLeavesTheAttemptReplayable() throws Exception
  {
    // every table of this backend is there except the one the catalog is kept in, which is the shape of an
    // installation whose tables predate the catalog
    final AtomicReference<String> catalogTable = new AtomicReference<>();
    final JDBCStorage storage = storageOverTablesThatAre(tableName -> !tableName.equals(catalogTable.get()));
    catalogTable.set(storage.getTableName(storage.getCatalogTree()));
    final AtomicInteger attempts = new AtomicInteger();

    storage.write(txn -> {
      txn.openTree(TREE, true);
      if (attempts.incrementAndGet() == 1)
      {
        throw new StorageRuntimeException(sql(0, "40001"));
      }
    });

    assertEquals(attempts.get(), 2, "the write that created and filled the catalog was not replayed");
    verify(statements, never()).executeUpdate();
    // and the catalog was filled, on the connection of its own: without this every assertion above holds of a
    // storage that enrolled nothing at all - an attempt issuing no statement is replayed the same way, and the
    // caller's connection is exactly as untouched. The row is the ANSI upsert of a plain mock, which tries an
    // update before an insert; the create is the table this fixture is missing
    verify(catalogConnection, atLeastOnce()).prepareStatement(startsWith("create table " + catalogTable.get()));
    verify(catalogConnection, atLeastOnce()).prepareStatement(startsWith("update " + catalogTable.get()));
    // committed where it is written, which is what keeps a row of an attempt that failed afterwards recorded
    verify(catalogConnection, atLeastOnce()).commit();
  }

  /**
   * A tree that had to be created did commit - the create table commits, and mysql and oracle commit before a DDL
   * statement of their own accord - so the attempt is out of the replay whatever the failure says: a
   * {@link WriteOperation} is only idempotent in the database, and {@code RootContainer.open()} replayed after the
@@ -745,24 +794,38 @@
  }

  /**
   * A storage whose pool hands out one connection of this test, over a catalog that either holds the table of
   * {@link #TREE} or does not. The connection is a mock of no recognized driver, which is how the engines that
   * guard their create index - and mssql, which has none - reach {@code openTree}.
   * A storage whose pool hands out one connection of this test, over a database that either holds the tables
   * this backend asks about or holds none of them - the table of {@link #TREE} and the table of the tree
   * catalog alike. The connection is a mock of no recognized driver, which is how the engines that guard their
   * create index - and mssql, which has none - reach {@code openTree}.
   */
  private JDBCStorage storageOverACatalogHolding(boolean theTable) throws Exception
  {
    return storageOverTablesThatAre(tableName -> theTable);
  }

  /**
   * The same, over a database holding exactly the tables the given rule accepts, and with a second connection
   * behind the pooled one: the tree catalog is written on a connection of its own, so that its rows commit
   * nothing of the caller's - which is the very thing {@code statements} is asserted on below.
   */
  private JDBCStorage storageOverTablesThatAre(Predicate<String> present) throws Exception
  {
    final Connection con = mock(Connection.class);
    final JDBCStorage storage = storageOver(con);
    final JDBCStorage storage = storageOver(con, catalogConnection());

    statements = mock(PreparedStatement.class);
    final String tableName = storage.getTableName(TREE);
    final DatabaseMetaData metaData = mock(DatabaseMetaData.class);
    // a result set of its own per call: the catalog is asked once per attempt, and a replayed attempt
    // reading a result set the previous one had already walked to its end would find no table there
    when(metaData.getTables(any(), any(), any(), any())).thenAnswer(invocation -> {
      // the name asked about and not the one table of a fixture: openTree() asks about the table of the tree
      // and about the table of the catalog, and answering the second with the name of the first would have
      // the catalog created over again on every attempt
      final String asked = (String) invocation.getArguments()[2];
      final ResultSet tables = mock(ResultSet.class);
      when(tables.next()).thenReturn(theTable, false);
      when(tables.getString("TABLE_NAME")).thenReturn(tableName);
      when(tables.next()).thenReturn(present.test(asked), false);
      when(tables.getString("TABLE_NAME")).thenReturn(asked);
      return tables;
    });

@@ -772,9 +835,37 @@
    return storage;
  }

  /** A connection of no rows at all, for a query this fixture has nothing to answer with. */
  private static ResultSet noRows() throws SQLException
  {
    final ResultSet rs = mock(ResultSet.class);
    when(rs.next()).thenReturn(false);
    return rs;
  }

  /**
   * A storage whose pool hands out one connection of the given engine, over a catalog holding the table of
   * {@link #TREE} and either holding its {@code k_} index or not. The index guard and the statement behind it
   * The connection the tree catalog of a storage of this test is written on: it opens one straight through the
   * driver, for the reason a stamp opens one of its own - the caller of openTree() is holding a pooled
   * connection already.
   */
  private Connection catalogConnection() throws Exception
  {
    final Connection con = mock(Connection.class);
    final PreparedStatement onIt = mock(PreparedStatement.class);
    final ResultSet empty = noRows(); // the read of what the catalog records: nothing was ever enrolled
    when(onIt.executeQuery()).thenReturn(empty);
    when(con.prepareStatement(anyString())).thenReturn(onIt);
    // the stamp of a tree name opens a connection of its own too, and a fixture that let it have this one
    // would have it issue the session statement of its dialect here
    when(con.createStatement()).thenThrow(new SQLException("no session statement in this test", "42000"));
    catalogConnection = con;
    return con;
  }

  /**
   * A storage whose pool hands out one connection of the given engine, over a database holding every table
   * this backend asks about - that of {@link #TREE} and that of its tree catalog - and either holding the
   * {@code k_} index of the first or not. The index guard and the statement behind it
   * are the branches {@code openTree()} takes per engine, and a mock of plain {@link Connection} reaches none
   * of them - so the name the mock ends up with is asserted here rather than assumed.
   */
@@ -787,11 +878,15 @@
        "a mock of " + engine.getSimpleName() + " reaches no " + engineName + " branch: "
            + JDBCStorage.driverNameOf(con));
    engineConnection = con;
    // the connections behind it answer the connects the pool does not make: the stamp of a tree name opens one
    // of its own, straight through the driver, since the caller of openTree() is holding a pooled connection
    final Connection[] answers = new Connection[behind.length + 1];
    // the connections behind it answer the connects the pool does not make: the tree catalog is read and
    // written on one of its own, straight through the driver, since the caller of openTree() is holding a
    // pooled connection already - and the stamp of a tree name opens one for the same reason. The catalog
    // comes first because openTree() opens the catalog before anything else and stamping its table is the
    // last thing that does, so a test naming a connection of its own names the one behind it
    final Connection[] answers = new Connection[behind.length + 2];
    answers[0] = con;
    System.arraycopy(behind, 0, answers, 1, behind.length);
    answers[1] = catalogConnection();
    System.arraycopy(behind, 0, answers, 2, behind.length);
    final JDBCStorage storage = storageOver(answers);

    statements = mock(PreparedStatement.class);
@@ -802,7 +897,8 @@
    when(metaData.getTables(any(), any(), any(), any())).thenAnswer(invocation -> {
      final ResultSet tables = mock(ResultSet.class);
      when(tables.next()).thenReturn(true, false);
      when(tables.getString("TABLE_NAME")).thenReturn(tableName);
      // the name asked about: openTree() asks about the table of the tree and about the table of the catalog
      when(tables.getString("TABLE_NAME")).thenReturn((String) invocation.getArguments()[2]);
      return tables;
    });
    when(metaData.getIndexInfo(any(), any(), any(), anyBoolean(), anyBoolean())).thenAnswer(invocation -> {
@@ -829,7 +925,11 @@
  private JDBCStorage storageOver(Connection... connections) throws Exception
  {
    final JDBCBackendCfg cfg = mock(JDBCBackendCfg.class);
    when(cfg.getDBDirectory()).thenReturn(StubDriver.PREFIX + pools.incrementAndGet());
    final int pool = pools.incrementAndGet();
    when(cfg.getDBDirectory()).thenReturn(StubDriver.PREFIX + pool);
    // the tree catalog of a backend is named after its id: a mock answering null for it would name every
    // storage of this class the same catalog, and the tables of these fixtures are named after that name
    when(cfg.getBackendId()).thenReturn("retry" + pool);
    final JDBCStorage storage = new JDBCStorage(cfg, null);
    storage.accessMode = AccessMode.READ_WRITE;
    stub.answerWith(connections);

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/PgSqlTestCase.java

@@ -11,17 +11,41 @@
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2025 3A Systems, LLC.
 * Copyright 2025-2026 3A Systems, LLC.
 */
package org.opends.server.backends.jdbc;

import org.opends.server.backends.pluggable.spi.AccessMode;
import org.opends.server.backends.pluggable.spi.TreeName;
import org.opends.server.backends.pluggable.spi.WriteOperation;
import org.opends.server.backends.pluggable.spi.WriteableTransaction;
import org.testcontainers.containers.JdbcDatabaseContainer;
import org.testcontainers.containers.PostgreSQLContainer;
import org.testng.annotations.Test;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertTrue;

//docker run --rm -it -p 5432:5432 -e POSTGRES_PASSWORD=password --name postgres postgres

@Test
/**
 * The class-level annotation governs the cases declared here, and it has to carry
 * {@code sequential = true} of its own: {@code TestListener.enforceTestClassTypeAndAnnotations()}
 * looks it up on the class declaring the case rather than on the one running it, so the
 * {@code @Test(groups = ..., sequential = true)} of {@code PluggableBackendImplTestCase} answers for
 * the inherited cases alone and a bare {@code @Test} here fails every case this class declares.
 * {@code OracleTestCase} carries it for the same reason.
 */
@Test(sequential = true)
public class PgSqlTestCase extends TestCase {

    @Override
@@ -48,4 +72,120 @@
        return "jdbc:postgresql://localhost:"+ ((container==null)?"5432":container.getMappedPort(5432))+"/database_name?user=postgres&password=password";
    }

    /** The schema put ahead of the one this suite's tables are in, for the case below and for nothing else. */
    private static final String AHEAD_ON_THE_PATH = "opendj_ahead";

    /**
     * A table of this backend is found where an unqualified statement of the same connection reaches it, and
     * not only in the schema that connection happens to work in.
     * <p>
     * PostgreSQL resolves an unqualified reference across the whole {@code search_path} while an unqualified
     * {@code create} lands in {@code current_schema()} alone, so the two are the same schema only as long as
     * nothing was put in front of the one the tables were made in. Adding a schema of its own to a role is the
     * standard remedy since PG15 took {@code CREATE} off {@code public}, and it makes them differ on an
     * installation whose tables are already there: the backend goes on reading and writing them unqualified,
     * and a lookup asking only about {@code current_schema()} would report every one of them absent. What that
     * would cost is this issue over again - the clear would drop nothing and say nothing, which is #888 - and
     * one thing worse besides: the next open would create a second, empty set of tables in the schema ahead,
     * and from that commit on they would shadow the populated ones for every later unqualified reference.
     * <p>
     * The connection string carries the path rather than a role being altered, because the pools of this
     * backend are keyed by it: a storage of another url is a storage of connections of its own, where an
     * {@code ALTER ROLE} would leave every connection already pooled resolving the way it always did.
     */
    @Test
    public void testAClearFindsATableOfAnotherSchemaOfTheSearchPath() throws Exception {
        final TreeName tree = new TreeName("testSearchPath", "tree");
        final String backendId = getBackendId() + "_searchPath";
        // the tables of an installation made before anything was put in front of the schema they are in
        final JDBCStorage created = new JDBCStorage(createBackendCfg(backendId), null);
        final String tableName = created.getTableName(tree);
        // the same backend, over connections resolving in a schema of its own first and in the one the
        // tables are in behind it: what they reach unqualified is unchanged, what they create is not
        final String aheadOfThem = getJdbcUrl() + "&currentSchema=" + AHEAD_ON_THE_PATH + ",public";
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(backendId, aheadOfThem), null);
        try {
            try {
                created.open(AccessMode.READ_WRITE);
                created.write(new WriteOperation() {
                    @Override
                    public void run(WriteableTransaction txn) throws Exception {
                        txn.openTree(tree, true);
                    }
                });
            } finally {
                created.close();
            }
            assertTrue(isExistsTable(tableName), "the case did not make the table it is about");

            try (final Connection con = DriverManager.getConnection(getJdbcUrl());
                 final Statement st = con.createStatement()) {
                st.execute("create schema if not exists " + AHEAD_ON_THE_PATH);
            }
            storage.open(AccessMode.READ_WRITE);
            try (final Connection con = DriverManager.getConnection(aheadOfThem)) {
                // the fixture is the whole of the case: without this the two schemas are the same one and
                // the assertions below hold of the version this case is about as well
                assertEquals(con.getSchema(), AHEAD_ON_THE_PATH,
                    "the connections of this storage do not work in the schema put ahead of the tables");
                assertNotEquals(con.getSchema(), "public", "the tables of this case are not in public after all");
            }

            assertTrue(storage.listTrees().contains(tree),
                "a tree whose table this connection reads unqualified was named by none of them");

            // the other half of what the narrowing decides, and the destructive one: openTree() creates a
            // table where its lookup answers that there is none, and an unqualified "create table" lands in
            // current_schema() - the schema ahead of the tables. A lookup asking about that schema alone
            // would answer no here and leave the populated table in public orphaned behind a second, empty
            // one, from this commit on. The clear below drops what the catalog names and would go on
            // passing while it happened, which is why this is asserted here rather than left to it
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(tree, true);
                }
            });
            assertFalse(isExistsTableInSchema(AHEAD_ON_THE_PATH, tableName),
                "the open created a second table in the schema ahead of the tables, shadowing the populated one");
            assertTrue(isExistsTableInSchema("public", tableName),
                "the open did not leave the populated table where it is");

            storage.removeStorageFiles();

            assertFalse(isExistsTable(tableName),
                "the clear left a table it reaches unqualified standing, for living in another schema of the search path");
        } finally {
            // the same backend id, so this clears what either half of the case created - including the
            // run where the clear under test drops nothing and the tables would otherwise be left for
            // whatever case of this class runs next
            clearQuietly(storage);
            clearQuietly(new JDBCStorage(createBackendCfg(backendId), null));
            try (final Connection con = DriverManager.getConnection(getJdbcUrl());
                 final Statement st = con.createStatement()) {
                st.execute("drop schema if exists " + AHEAD_ON_THE_PATH + " cascade");
            }
        }
    }

    /**
     * Whether the table is in that one schema, which is the question the case above asks and the one
     * {@code TestCase.isExistsTable} cannot answer: it walks every schema the connection can see, so a
     * table created in the wrong one of the two reads there exactly like a table created in the right
     * one. Asked of {@code information_schema} with the schema and the name bound rather than through
     * {@code getTables()}, whose schema is a pattern - {@code opendj_ahead} would match a schema named
     * {@code opendjXahead} as readily, {@code _} being a single-character wildcard there.
     */
    private boolean isExistsTableInSchema(String schema, String tableName) throws SQLException {
        try (final Connection con = DriverManager.getConnection(getJdbcUrl());
             final PreparedStatement st = con.prepareStatement(
                 "select 1 from information_schema.tables where table_schema=? and lower(table_name)=lower(?)")) {
            st.setString(1, schema);
            st.setString(2, tableName);
            try (final ResultSet rs = st.executeQuery()) {
                return rs.next();
            }
        }
    }

}

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java

@@ -15,8 +15,10 @@
 */
package org.opends.server.backends.jdbc;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.ldap.ByteStringBuilder;
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.backends.pluggable.PluggableBackendImplTestCase;
import org.opends.server.backends.pluggable.spi.AccessMode;
@@ -41,12 +43,16 @@
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Properties;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -121,12 +127,68 @@

    @Override
    protected JDBCBackendCfg createBackendCfg() {
        return createBackendCfg(getBackendId());
    }

    /**
     * A configuration of another backend on the database of this suite: backends sharing one database
     * URL is a configuration nothing forbids, and what one of them clears must be its own tables.
     */
    protected JDBCBackendCfg createBackendCfg(String backendId) {
        JDBCBackendCfg backendCfg = mockCfg(JDBCBackendCfg.class);
        when(backendCfg.getBackendId()).thenReturn(getBackendId());
        when(backendCfg.getBackendId()).thenReturn(backendId);
        when(backendCfg.getDBDirectory()).thenReturn(getJdbcUrl());
        return backendCfg;
    }

    /**
     * The same, reached over a connection string of the caller's own: the pools of this backend are
     * keyed by it, so a case wanting connections established differently - in another schema of the
     * search path, say - asks for them by asking for another url.
     */
    protected JDBCBackendCfg createBackendCfg(String backendId, String jdbcUrl) {
        final JDBCBackendCfg backendCfg = createBackendCfg(backendId);
        when(backendCfg.getDBDirectory()).thenReturn(jdbcUrl);
        return backendCfg;
    }

    /**
     * The same, serving the given base DN: what a clear compares the tree stamp of a table against
     * when it says whether the table is this backend's own or another's (#866).
     */
    protected JDBCBackendCfg createBackendCfg(String backendId, DN baseDN) {
        final JDBCBackendCfg backendCfg = createBackendCfg(backendId);
        final TreeSet<DN> baseDNs = new TreeSet<>();
        baseDNs.add(baseDN);
        when(backendCfg.getBaseDN()).thenReturn(baseDNs);
        return backendCfg;
    }

    /** Asked of the database itself, by listing its tables, so that no folding rule of the backend is trusted here. */
    protected boolean isExistsTable(String tableName) throws SQLException {
        try (final Connection con = DriverManager.getConnection(getJdbcUrl())) {
            return isExistingTable(con, tableName);
        }
    }

    /** Drops a table behind the back of the storage that owns it, which no code path of the backend does. */
    private void dropTableBehindTheBackend(String tableName) throws SQLException {
        try (final Connection con = DriverManager.getConnection(getJdbcUrl());
             final Statement st = con.createStatement()) {
            st.execute("drop table " + tableName);
        }
    }

    /** Clears a backend of a test without letting the failure of the clear replace the failure being reported. */
    protected static void clearQuietly(JDBCStorage storage) {
        try {
            storage.removeStorageFiles();
        } catch (Exception ignored) {
        } finally {
            storage.close();
        }
    }

    @AfterClass
    @Override
    public void cleanUp() throws Exception {
@@ -314,11 +376,16 @@
    }

    /**
     * Reading a tree must not enrol it in the storage's tree map: removeStorageFiles() drops every
     * table that map names, and the compressed schema reads the tree its definitions used to be
     * shared under - which on a shared database is another backend's to keep (#873). Asking whether
     * the tree is there is only the first of those reads: the migration counts it and copies it out
     * too, so one guarded statement would not be enough.
     * Reading a tree must not put it up for removal: a clear drops what the catalog of the backend
     * names (#888), and the compressed schema reads the tree its definitions used to be shared under
     * - which on a shared database is another backend's to keep (#873). Asking whether the tree is
     * there is only the first of those reads: the migration counts it and copies it out too, so one
     * guarded statement would not be enough.
     * <p>
     * The two storages are two backends and not one addressing the same database, which is what the
     * case is about: what a backend owns is recorded in a catalog named after its backend id and
     * outlives the process that opened the tree, so a second storage of the same id would be shown
     * the tree its own earlier open had enrolled - and would be right to be.
     */
    @Test
    public void testProbingATreeDoesNotPutItUpForRemoval() throws Exception {
@@ -334,9 +401,9 @@
        });
        owner.close();

        // a second storage on the same database, which never opened that tree - the shape of two
        // a second backend on the same database, which never opened that tree - the shape of two
        // backends addressing one database
        final JDBCStorage other = new JDBCStorage(createBackendCfg(), null);
        final JDBCStorage other = new JDBCStorage(createBackendCfg(getBackendId() + "_probe"), null);
        try {
            other.open(AccessMode.READ_WRITE);
            other.read(new ReadOperation<Void>() {
@@ -399,7 +466,11 @@
        });
        owner.close();

        // a storage that never opened that tree, so nothing but the delete can enrol it
        // a second storage of the same backend, which never opened that tree itself: the delete takes
        // the enrolling name and the table it writes to is the one the tree names. What puts a tree up
        // for removal is the row its backend's catalog holds (#888) - written by the openTree above and
        // outliving the storage that made it - so this asserts the listing of a backend and not a
        // side effect of the statement, which is what a listing of a catalog can assert
        final JDBCStorage other = new JDBCStorage(createBackendCfg(), null);
        try {
            other.open(AccessMode.READ_WRITE);
@@ -1109,7 +1180,10 @@
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.put(written, key(1), value(1)); // pending in this transaction...
                    // pending in this transaction, and it has to still be pending when the stamp is
                    // attempted: both trees were enrolled by the storage above, so the openTree below
                    // records nothing in the catalog and commits nothing of what is written here
                    txn.put(written, key(1), value(1));
                    txn.openTree(stamped, true); // ...while the comment machinery fails
                }
            });
@@ -1120,9 +1194,12 @@
                    return null;
                }
            });
            // the failure is remembered: an unstampable table is not asked again while this backend is open
            // the failure is remembered: an unstampable table is not asked again while this backend is open.
            // Counted from what the open itself attempted rather than from one: the open stamps the tree and
            // the catalog of the backend, and how many tables an open has to stamp is not what this is about
            final int attemptsOfTheOpen = stampAttempts.get();
            assertEquals(storage.commentTable(stamped, dialect()), JDBCStorage.CommentResult.FAILED);
            assertEquals(stampAttempts.get(), 1, "a failed stamp was reissued");
            assertEquals(stampAttempts.get(), attemptsOfTheOpen, "a failed stamp was reissued");
        } finally {
            try {
                storage.write(new WriteOperation() {
@@ -1832,4 +1909,930 @@
            storage.close();
        }
    }

    /**
     * removeStorageFiles() has to clear a backend this process has never opened: offline import-ldif
     * configures the backend and calls it before anything opens the root container, so answering from
     * the trees this process happens to have touched dropped nothing at all - an offline
     * "import-ldif --clearBackend" cleared a JDBC backend of nothing (#888).
     */
    @Test
    public void testABackendIsClearedByAProcessThatNeverOpenedIt() throws Exception {
        final TreeName tree = new TreeName("testOfflineClear", "tree");
        // the neighbour serves a base DN of its own, so that its table is one it reports as its own:
        // what this case asserts of the clear next door is then an absence and not a vacuity
        final DN neighbourBaseDN = DN.valueOf("dc=offline-clear-neighbour,dc=com");
        final TreeName neighbourTree = new TreeName(neighbourBaseDN.toNormalizedUrlSafeString(), "tree");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_cleared"), null);
        final JDBCStorage neighbour =
            new JDBCStorage(createBackendCfg(getBackendId() + "_neighbour", neighbourBaseDN), null);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(tree, true);
                    txn.put(tree, key(1), value(1));
                }
            });
            neighbour.open(AccessMode.READ_WRITE);
            neighbour.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(neighbourTree, true);
                    txn.put(neighbourTree, key(1), value(1));
                }
            });
        } catch (Exception e) {
            // the clears of the case below are reached by no failure of this half, and nothing but
            // @BeforeClass ever drops what it leaves behind
            clearQuietly(storage);
            clearQuietly(neighbour);
            throw e;
        } finally {
            storage.close();
            neighbour.close();
        }

        // configured and never opened, nothing touched: what BackendImpl.importLDIF holds offline
        final JDBCStorage offline = new JDBCStorage(createBackendCfg(getBackendId() + "_cleared"), null);
        try {
            assertTrue(offline.listTrees().contains(tree),
                "the tree of a backend this process never opened has to be named by its catalog");

            offline.removeStorageFiles();

            assertFalse(isExistsTable(offline.getTableName(tree)), "the table of the tree survived the clear");
            assertFalse(isExistsTable(offline.getTableName(offline.getCatalogTree())), "the catalog survived the clear");
            final Set<TreeName> cleared = offline.listTrees();
            assertFalse(cleared.contains(tree), "a cleared backend still names its tree");
            assertFalse(cleared.contains(offline.getCatalogTree()), "a cleared backend still names its catalog");
            // the neighbour is named by a catalog of its own: what one backend clears is never another's
            assertTrue(isExistsTable(neighbour.getTableName(neighbourTree)),
                "the clear of one backend dropped the table of another backend of the same database");
            // nor does it report another backend's tables as tables of its own: a table is named after
            // the hash of its tree name and says nothing about whose it is, but it is stamped with that
            // tree name (#866), and the neighbour's trees are trees of no base DN this backend serves
            assertReportsNothingOf(offline, neighbour, neighbourTree);
        } finally {
            // in a finally of their own: a failed assertion above must not leave the tables of either
            // backend behind for the rest of the class, which nothing but @BeforeClass ever drops
            clearQuietly(neighbour);
            clearQuietly(offline);
        }
    }

    /**
     * A dropped tree has to leave the catalog together with its table: a row outliving its table
     * would make backendstat name a tree that is not there, and would put a table that is already
     * gone up for removal (#888).
     */
    @Test
    public void testADeletedTreeIsNoLongerNamedByTheCatalog() throws Exception {
        final TreeName kept = new TreeName("testCatalogDelete", "kept");
        final TreeName dropped = new TreeName("testCatalogDelete", "dropped");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_deleted"), null);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(kept, true);
                    txn.openTree(dropped, true);
                }
            });
            final Set<TreeName> opened = storage.listTrees();
            assertTrue(opened.contains(kept) && opened.contains(dropped), "an opened tree is not named by the catalog");

            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.deleteTree(dropped);
                }
            });

            final Set<TreeName> remaining = storage.listTrees();
            assertTrue(remaining.contains(kept), "the catalog forgot a tree that is still there");
            assertFalse(remaining.contains(dropped), "the catalog still names a tree that was deleted");
            // and the removal that follows must not stumble over the tree it no longer names
            storage.removeStorageFiles();
            assertFalse(isExistsTable(storage.getTableName(kept)), "the table of the tree survived the clear");
        } finally {
            clearQuietly(storage);
        }
    }

    /**
     * A row of the catalog whose table is not there any more must not fail the clear, and must not
     * stop it dropping the rest. Nothing of the backend leaves such a row behind - deleteTree() takes
     * it out in the commit that drops the table - but a table dropped by hand, or a catalog restored
     * from a backup older than the database, leaves exactly this (#888).
     */
    @Test
    public void testAClearSkipsACatalogRowWhoseTableIsGone() throws Exception {
        final TreeName kept = new TreeName("testStaleCatalogRow", "kept");
        final TreeName vanished = new TreeName("testStaleCatalogRow", "vanished");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_stale"), null);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(kept, true);
                    txn.openTree(vanished, true);
                }
            });
            dropTableBehindTheBackend(storage.getTableName(vanished));
            assertTrue(storage.listTrees().contains(vanished),
                "the catalog was expected to go on naming the tree whose table was dropped behind its back");

            storage.removeStorageFiles();

            assertFalse(isExistsTable(storage.getTableName(kept)),
                "a row of the catalog whose table is gone stopped the clear dropping the rest");
            assertFalse(isExistsTable(storage.getTableName(storage.getCatalogTree())), "the catalog survived the clear");
        } finally {
            clearQuietly(storage);
        }
    }

    /**
     * Naming a tree in order to read it must never put it up for removal: the tree read may be held
     * by another backend of the same database, which nothing forbids (#873). Only
     * openTree(createOnDemand) enrols.
     */
    @Test
    public void testReadingATreeDoesNotPutItUpForRemoval() throws Exception {
        final TreeName owned = new TreeName("testReadDoesNotEnrol", "owned");
        final TreeName foreign = new TreeName("testReadDoesNotEnrolForeign", "tree");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_reader"), null);
        final JDBCStorage owner = new JDBCStorage(createBackendCfg(getBackendId() + "_owner"), null);
        try {
            owner.open(AccessMode.READ_WRITE);
            owner.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(foreign, true);
                    txn.put(foreign, key(1), value(1));
                }
            });

            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(owned, true); // the catalog of this backend comes into being here
                    txn.openTree(foreign, false); // read, not owned
                }
            });
            assertEquals(storage.read(new ReadOperation<ByteString>() {
                @Override
                public ByteString run(ReadableTransaction txn) throws Exception {
                    return txn.read(foreign, key(1));
                }
            }), value(1), "the tree of the other backend could not be read");

            assertFalse(storage.listTrees().contains(foreign), "reading a tree enrolled it in the catalog");
            storage.removeStorageFiles();
            assertTrue(isExistsTable(owner.getTableName(foreign)),
                "the clear dropped a tree this backend had only read");
            assertFalse(isExistsTable(storage.getTableName(owned)), "the table of the backend's own tree survived the clear");
        } finally {
            clearQuietly(owner);
            clearQuietly(storage);
        }
    }

    /**
     * The compressed schema trees named from a literal carry no backend qualifier, so on a database
     * addressed by several backends they are the same pair for all of them: a clear must leave them
     * where they lie (#881). A tool asking a backend what trees it holds has to be shown them all the
     * same, which is what keeps them out of the catalog and inside listTrees().
     */
    @Test
    public void testTheSharedCompressedSchemaTreesAreNamedButNeverCleared() throws Exception {
        final TreeName owned = new TreeName("testSharedCompressedSchema", "owned");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_schema"), null);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(owned, true);
                    // opened, never written to: since #881 no backend of this class makes the literal-named
                    // pair, so this openTree is what creates these two tables - and the finally below is what
                    // removes them again, a clear being required to leave them exactly where they lie
                    for (final TreeName shared : JDBCStorage.SHARED_COMPRESSED_SCHEMA_TREES) {
                        txn.openTree(shared, true);
                    }
                }
            });
            // both of them: the pair is a hand-copy of two privates of PersistentCompressedSchema, and
            // a literal naming a tree that does not exist would go unseen if one of them were never asked
            // for - the tree it names would be neither shown by listTrees() nor spared by a clear
            final Set<TreeName> named = storage.listTrees();
            for (final TreeName shared : JDBCStorage.SHARED_COMPRESSED_SCHEMA_TREES) {
                assertTrue(named.contains(shared),
                    "a tool asking this backend for its trees was not shown " + shared);
            }

            storage.removeStorageFiles();

            for (final TreeName shared : JDBCStorage.SHARED_COMPRESSED_SCHEMA_TREES) {
                assertTrue(isExistsTable(JDBCStorage.toTableName(shared)),
                    "the clear dropped " + shared + ", which another backend of this database may be the only owner of");
            }
            assertFalse(isExistsTable(storage.getTableName(owned)), "the table of the backend's own tree survived the clear");
        } finally {
            // the pair is dropped by hand here, and by nothing of the backend: a clear must leave it
            // where it lies, which is the whole of what this case asserts. It is this case's to remove
            // because it is this case that made it - since #881 each backend keeps its definitions in a
            // pair of its own, so the literal-named pair belongs to no backend of this class any more
            // and the openTree above is what created these two tables. Left standing they would be a
            // legacy pair this database does not have, which
            // testCompressedSchemaTableIsQualifiedByBackendId asserts about and TestNG may run after
            // this case as easily as before it
            clearQuietly(storage);
            for (final TreeName shared : JDBCStorage.SHARED_COMPRESSED_SCHEMA_TREES) {
                try {
                    dropTableBehindTheBackend(JDBCStorage.toTableName(shared));
                } catch (SQLException ignored) { // a case that failed before it made them leaves none to drop
                }
            }
        }
    }

    /**
     * The row of a deleted tree must not be left to the enclosing transaction: a terminal failure
     * later in it - write() replays a class 40 conflict and rethrows everything else - would roll the
     * row back over a table that is already gone, and nothing would put it right, a deleted tree not
     * being opened again (#888).
     */
    @Test
    public void testADeletedTreeStaysOutOfTheCatalogWhenItsTransactionFails() throws Exception {
        final TreeName kept = new TreeName("testCatalogDeleteRollback", "kept");
        final TreeName deleted = new TreeName("testCatalogDeleteRollback", "deleted");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_rollback"), null);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(kept, true);
                    txn.openTree(deleted, true);
                }
            });

            try {
                storage.write(new WriteOperation() {
                    @Override
                    public void run(WriteableTransaction txn) throws Exception {
                        txn.deleteTree(deleted);
                        // terminal, and no conflict for write() to replay: everything this transaction
                        // still owes goes back, and the row of the deleted tree must not be part of it
                        throw new IllegalStateException("the transaction of a deleteTree failed");
                    }
                });
                fail("the write was expected to fail");
            } catch (Exception expected) {
                // what the case is about is what the failure left behind
            }

            assertFalse(isExistsTable(storage.getTableName(deleted)), "the failed transaction brought a dropped table back");
            final Set<TreeName> remaining = storage.listTrees();
            assertFalse(remaining.contains(deleted),
                "the catalog names a tree whose table the failed transaction left dropped");
            assertTrue(remaining.contains(kept), "the catalog forgot a tree that is still there");
        } finally {
            clearQuietly(storage);
        }
    }

    /**
     * The same, for the branch a deleteTree takes when the table is not there any more: nothing is
     * dropped, so there is no commit of a drop for the row to be carried out of the catalog by, and
     * the commit the delete is given on the catalog's own connection is the whole of what takes it
     * out. Left to the enclosing transaction, the row would go back with it and the catalog would name
     * a tree with no table for good - the state a clear can only skip and report, never repair (#888).
     */
    @Test
    public void testADeletedTreeStaysOutOfTheCatalogWhenItsTableIsAlreadyGone() throws Exception {
        final TreeName kept = new TreeName("testCatalogDeleteNoTable", "kept");
        final TreeName deleted = new TreeName("testCatalogDeleteNoTable", "deleted");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_noTable"), null);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(kept, true);
                    txn.openTree(deleted, true);
                }
            });
            // what an interrupted change of an earlier run leaves: a row of the catalog naming a table
            // that is not there any more. The deleteTree below therefore drops nothing at all
            dropTableBehindTheBackend(storage.getTableName(deleted));

            try {
                storage.write(new WriteOperation() {
                    @Override
                    public void run(WriteableTransaction txn) throws Exception {
                        txn.deleteTree(deleted);
                        // terminal, and no conflict for write() to replay: everything this transaction
                        // still owes goes back, and the row of the deleted tree must not be part of it
                        throw new IllegalStateException("the transaction of a deleteTree failed");
                    }
                });
                fail("the write was expected to fail");
            } catch (Exception expected) {
                // what the case is about is what the failure left behind
            }

            final Set<TreeName> remaining = storage.listTrees();
            assertFalse(remaining.contains(deleted),
                "the catalog names a tree whose table was already gone when it was deleted");
            assertTrue(remaining.contains(kept), "the catalog forgot a tree that is still there");
        } finally {
            clearQuietly(storage);
        }
    }

    /**
     * The row of a tree whose table is already standing must not be left to the enclosing transaction
     * either. That is the open which fills the catalog of a backend upgraded from a version keeping
     * none: it creates no table, so nothing else of openTree() commits anything, and a transaction
     * failing after the enrolment would take the whole of it back - leaving a backend whose tables
     * are named by no catalog and whose next clear therefore drops nothing at all (#888).
     * <p>
     * The row is written and committed on a connection of the catalog's own, so this holds on every
     * engine for the same reason: nothing the caller's transaction does - or fails to do - reaches it.
     * On the branch before this one the row rode the caller's connection, and the case was green on
     * postgres for a reason of that engine alone (openTree() asks there for the cursor index of every
     * tree on every open and commits that, carrying the row with it) while the other three lost it.
     */
    @Test
    public void testAReopenedTreeStaysInTheCatalogWhenItsTransactionFails() throws Exception {
        final TreeName tree = new TreeName("testCatalogEnrolRollback", "tree");
        final JDBCStorage setUp = new JDBCStorage(createBackendCfg(getBackendId() + "_enrol"), null);
        try { // the tables of the backend, made by a storage that then goes away
            setUp.open(AccessMode.READ_WRITE);
            setUp.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(tree, true);
                }
            });
        } finally {
            setUp.close();
        }
        // and the rest of what an installation upgraded to a version keeping a catalog holds: a
        // catalog naming none of those tables
        emptyTheCatalog(setUp.getTableName(setUp.getCatalogTree()));

        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_enrol"), null);
        try {
            storage.open(AccessMode.READ_WRITE);
            assertFalse(storage.listTrees().contains(tree), "the catalog of the case was not emptied");

            try {
                storage.write(new WriteOperation() {
                    @Override
                    public void run(WriteableTransaction txn) throws Exception {
                        // the table is there, so this open creates none: the enrolment is the only thing
                        // this transaction has written when it fails
                        txn.openTree(tree, true);
                        // terminal, and no conflict for write() to replay: everything this transaction
                        // still owes goes back, and the row naming a standing table must not be part of it
                        throw new IllegalStateException("the transaction of an openTree failed");
                    }
                });
                fail("the write was expected to fail");
            } catch (Exception expected) {
                // what the case is about is what the failure left behind
            }

            assertTrue(storage.listTrees().contains(tree),
                "the catalog forgot a tree whose table is standing: a clear of this backend would drop nothing");
        } finally {
            clearQuietly(storage);
        }
    }

    /**
     * A tree the catalog already records at the table this version records it at is not enrolled
     * again by an open: the row would be the row that is already there. A row recording any other
     * table is, though - a removal drops the table the row records, so a row naming one this backend
     * would not create leaves the real table standing, named by nothing and dropped by no clear ever
     * after. Which of the two a row is has to be decided by what it records and not by its presence.
     */
    @Test
    public void testARowRecordingAnotherTableIsEnrolledAgain() throws Exception {
        final TreeName tree = new TreeName("testCatalogStaleRow", "tree");
        final JDBCStorage setUp = new JDBCStorage(createBackendCfg(getBackendId() + "_staleRow"), null);
        try { // the table and its row, by a storage that then goes away
            setUp.open(AccessMode.READ_WRITE);
            setUp.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(tree, true);
                }
            });
        } finally {
            setUp.close();
        }
        final String catalogTable = setUp.getTableName(setUp.getCatalogTree());
        // what a version naming its tables otherwise would have left: a row of the right tree
        // recording a table this one would never create
        recordAnotherTable(catalogTable, "opendj_00000000000000000000000000000000000000000000000000000000");

        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_staleRow"), null);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(tree, true);
                }
            });

            try (final Connection con = DriverManager.getConnection(getJdbcUrl())) {
                assertEquals(
                    storage.catalogTables(con, JDBCStorage.TableScope.of(storage, con)).get(tree),
                    storage.getTableName(tree),
                    "a row recording a table this backend does not hold was left as it was: its tree is named at a table no clear can drop");
            }
        } finally {
            clearQuietly(storage);
        }
    }

    /**
     * A clear which removed no tree of its backend says why, and says it where the one table it did
     * drop was its own catalog: a catalog standing over rows that name nothing - the state a backup
     * restored beside older tables leaves - is one drop and no tree removed, which is the outcome of
     * #888 exactly and not a clear that did something.
     * <p>
     * Decided on the drops of trees and not on every drop for that reason. Counted the other way the
     * line is silent here, since dropping the catalog makes the count one.
     */
    @Test
    public void testAClearWhichRemovedNoTreeSaysWhyEvenWhereItDroppedItsCatalog() throws Exception {
        final DN baseDN = DN.valueOf("dc=clear-catalog-only,dc=com");
        final TreeName owned = new TreeName(baseDN.toNormalizedUrlSafeString(), "id2entry");
        final ReportingStorage storage =
            new ReportingStorage(createBackendCfg(getBackendId() + "_catalogOnly", baseDN));
        final String catalogTable = storage.getTableName(storage.getCatalogTree());
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(owned, true);
                }
            });
            // the catalog table is there and names nothing, so the clear below has exactly one table to
            // drop - its own - and leaves the tree standing, named by nothing
            emptyTheCatalog(catalogTable);
            storage.close();

            storage.removeStorageFiles();

            assertFalse(isExistsTable(catalogTable), "the clear left its own catalog table standing");
            assertTrue(isExistsTable(storage.getTableName(owned)),
                "a table named by no catalog was dropped: nothing may be dropped that cannot be attributed");
            storage.assertReported("a clear which dropped its catalog and removed no tree of the backend"
                    + " said nothing about why, which is the silence of #888",
                "removed no tree of this backend", "has to be started once");
        } finally {
            clearQuietly(storage);
            // left standing on purpose above: its catalog is gone, so no clear of this backend names it
            dropTableIfExists(storage.getTableName(owned));
        }
    }

    /**
     * A row recording a name outside the namespace this backend names its tables in is passed over
     * rather than reaching a {@code drop table} built from a value read back out of a table - and the
     * clear accounts for it, no other line of its report being able to: what such a row records is
     * outside the {@code opendj} names the scan of what a clear left standing walks, and is dropped
     * by nothing. The row is not there to be read again either - the catalog names itself last, so
     * the clear drops that table with the row still in it - which is why the line is asserted here
     * along with the drop: it is the only surviving copy of what the row said.
     * <p>
     * Nothing this version writes makes such a row, which is why the case makes one by hand.
     */
    @Test
    public void testAClearAccountsForACatalogRowItCannotActOn() throws Exception {
        final TreeName tree = new TreeName("testCatalogForeignRow", "tree");
        final ReportingStorage storage = new ReportingStorage(createBackendCfg(getBackendId() + "_foreignRow"));
        final String tableName = storage.getTableName(tree);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(tree, true);
                }
            });
            final String catalogTable = storage.getTableName(storage.getCatalogTree());
            recordAnotherTable(catalogTable, "a_table_of_something_else");

            try (final Connection con = DriverManager.getConnection(getJdbcUrl())) {
                final List<String> skipped = new ArrayList<>();
                assertFalse(storage.readCatalogRows(con, catalogTable, skipped).containsKey(tree),
                    "a row recording a name no table of this backend goes by was read as a tree to drop");
                assertEquals(skipped.size(), 1, "the row the read passed over was not described to its caller: " + skipped);
                assertTrue(skipped.get(0).contains("a_table_of_something_else"),
                    "what the row records is named by nothing the clear could report: " + skipped);
            }

            // the clear still drops what it can: the catalog itself, which it names last
            storage.removeStorageFiles();
            assertTrue(isExistsTable(tableName),
                "the clear dropped the table of a tree its catalog names at another name than that table's");
            assertFalse(isExistsTable(catalogTable),
                "the clear left its own catalog table standing, so the row it passed over is still readable"
                    + " and the line reporting it is not the last copy of what it said");
            // the report itself and not the read behind it: reportSkippedRows() writes to nothing else,
            // so both of its call sites could be deleted and every assertion above would still hold
            storage.assertReported("the row the clear could not act on was reported by no line of it",
                "a_table_of_something_else", "passed over");
        } finally {
            clearQuietly(storage);
            // left standing on purpose above, so this case removes it rather than the next one meeting it
            dropTableIfExists(tableName);
        }
    }

    /**
     * A clear drops the table its catalog records for a tree, not one it derives again from the tree
     * name, so that a removal drops what was enrolled even if the naming of tables were ever to
     * change. A row recording no table at all - all a version recording the name alone would have
     * left - falls back to the derived name rather than naming nothing.
     */
    @Test
    public void testAClearDropsTheTableTheCatalogRecords() throws Exception {
        final TreeName tree = new TreeName("testCatalogValue", "tree");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_value"), null);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(tree, true);
                }
            });
            try (final Connection con = DriverManager.getConnection(getJdbcUrl())) {
                // asked the way a clear asks it, narrowed to where an unqualified name of the connection
                // resolves: what the removal reads is this and not a lookup of a shape of its own
                final Map<TreeName, String> recorded =
                    storage.catalogTables(con, JDBCStorage.TableScope.of(storage, con));
                assertEquals(recorded.get(tree), storage.getTableName(tree),
                    "the catalog does not record the table holding the tree its row names");

                emptyTheRecordedTableNames(storage.getTableName(storage.getCatalogTree()));
                assertEquals(storage.catalogTables(con, JDBCStorage.TableScope.of(storage, con)).get(tree),
                    storage.getTableName(tree),
                    "a row recording no table name did not fall back to the name derived from the tree");
            }

            storage.removeStorageFiles();

            assertFalse(isExistsTable(storage.getTableName(tree)), "the table the catalog named survived the clear");
        } finally {
            clearQuietly(storage);
        }
    }

    /**
     * A clear drops the catalog last, after every tree it names: what names the trees has to outlive
     * them. Dropping a table is DDL, which mysql and oracle commit as they go, so a clear that fails
     * halfway leaves a catalog still naming what is left - and the next attempt finishes it - where one
     * that had dropped the catalog first would leave tables nothing names any more and no clear could
     * ever reach.
     * <p>
     * Taken from the drops themselves and not from the map the loop walks: the map is built with the
     * catalog put last by hand, so an assertion on it would hold of any loop at all - one that sorted
     * the keys, or copied them into a HashSet, included.
     */
    @Test
    public void testAClearDropsTheCatalogAfterEveryTreeItNames() throws Exception {
        final TreeName first = new TreeName("testCatalogDropOrder", "first");
        final TreeName second = new TreeName("testCatalogDropOrder", "second");
        final List<String> order = new ArrayList<>();
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_order"), null) {
            @Override
            void dropTable(Connection con, String tableName) throws SQLException {
                order.add(tableName);
                super.dropTable(con, tableName);
            }
        };
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(first, true);
                    txn.openTree(second, true);
                }
            });

            storage.removeStorageFiles();

            final String catalogTable = storage.getTableName(storage.getCatalogTree());
            assertTrue(order.contains(storage.getTableName(first)) && order.contains(storage.getTableName(second)),
                "the clear did not drop the tables of the trees its catalog names: " + order);
            assertEquals(order.get(order.size() - 1), catalogTable,
                "the clear dropped the catalog before a tree it names, which no later clear could reach: " + order);
            assertEquals(order.indexOf(catalogTable), order.size() - 1,
                "the catalog was dropped more than once: " + order);
        } finally {
            clearQuietly(storage);
        }
    }

    /**
     * What a clear leaves standing it reports, and it reports it as what it is: a table stamped with a
     * tree of a base DN this backend serves is its own and can be removed by hand, while a table of a
     * backend sharing this database (#873) is that backend's business and no part of this outcome.
     * Told apart by the stamp of #866 and by nothing else - a table name is a bare hash.
     */
    @Test
    public void testAClearReportsTheTablesItCanAttributeToThisBackend() throws Exception {
        final DN baseDN = DN.valueOf("dc=clear-report,dc=com");
        final TreeName owned = new TreeName(baseDN.toNormalizedUrlSafeString(), "id2entry");
        // a base DN of its own, so that the neighbour is a backend that reports this table as its own:
        // what this case asserts about the clear of the other one is then an absence and not a vacuity
        final DN neighbourBaseDN = DN.valueOf("dc=clear-report-neighbour,dc=com");
        final TreeName neighbourTree = new TreeName(neighbourBaseDN.toNormalizedUrlSafeString(), "id2entry");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(getBackendId() + "_reported", baseDN), null);
        final JDBCStorage neighbour =
            new JDBCStorage(createBackendCfg(getBackendId() + "_reportedNeighbour", neighbourBaseDN), null);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(owned, true);
                }
            });
            neighbour.open(AccessMode.READ_WRITE);
            neighbour.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(neighbourTree, true);
                }
            });
            // the state of a backend upgraded from a version keeping no catalog: its tables are there
            // and nothing names them, so the clear that follows drops nothing at all
            dropTableBehindTheBackend(storage.getTableName(storage.getCatalogTree()));
            storage.close();

            storage.removeStorageFiles();

            assertTrue(isExistsTable(storage.getTableName(owned)),
                "a table named by no catalog was dropped: nothing may be dropped that cannot be attributed");
            try (final Connection con = DriverManager.getConnection(getJdbcUrl())) {
                // asked the way a clear asks it, through the same normalisation: a driver naming its
                // catalog with an empty string names no catalog, and a metadata pattern reads that as
                // "the tables belonging to no catalog at all", which would answer nothing
                final JDBCStorage.ClearLeftovers leftovers =
                    storage.leftoverTables(con, JDBCStorage.TableScope.of(storage, con));
                assertNotNull(leftovers, "the database would not say which tables the clear left standing");
                assertTrue(leftovers.ours.toString().toLowerCase().contains(storage.getTableName(owned).toLowerCase()),
                    "a table of a base DN this backend serves was not reported as its own: " + leftovers.ours);
                assertFalse(leftovers.unattributed.toString().toLowerCase().contains(storage.getTableName(owned).toLowerCase()),
                    "a table this backend can name was reported as attributable to nobody: " + leftovers.unattributed);
                assertTrue(leftovers.unreadable.isEmpty(),
                    "the stamp of a table this database does give up was reported as unreadable: " + leftovers.unreadable);
            }
            assertReportsNothingOf(storage, neighbour, neighbourTree);
        } finally {
            clearQuietly(neighbour);
            // the catalog of this one is gone, so its clear names nothing: the table it left standing on
            // purpose is dropped here by hand, as the report says such a table has to be
            clearQuietly(storage);
            dropTableIfExists(storage.getTableName(owned));
        }
    }

    /**
     * Asserts that the clear of one backend says nothing whatsoever about the tables of another - and
     * that the silence is one about tables the scan does reach: the backend those tables belong to is
     * asked the same question and reports them as its own, so an absence here is a decision and not a
     * scan that enumerated nothing.
     */
    private void assertReportsNothingOf(JDBCStorage cleared, JDBCStorage other, TreeName otherTree) throws SQLException {
        try (final Connection con = DriverManager.getConnection(getJdbcUrl())) {
            final JDBCStorage.ClearLeftovers leftovers =
                cleared.leftoverTables(con, JDBCStorage.TableScope.of(cleared, con));
            assertNotNull(leftovers, "the database would not say which tables the clear left standing");
            final String reported =
                (leftovers.ours + " " + leftovers.unattributed + " " + leftovers.unreadable).toLowerCase();
            assertFalse(reported.contains(other.getTableName(otherTree).toLowerCase()),
                "the clear of one backend reported the table of another: " + reported);
            assertFalse(reported.contains(other.getTableName(other.getCatalogTree()).toLowerCase()),
                "the clear of one backend reported the catalog of another: " + reported);

            final JDBCStorage.ClearLeftovers theirs =
                other.leftoverTables(con, JDBCStorage.TableScope.of(other, con));
            assertNotNull(theirs, "the database would not say which tables the neighbour is holding");
            assertTrue(theirs.ours.toString().toLowerCase().contains(other.getTableName(otherTree).toLowerCase()),
                "the table left unreported is one the scan does not reach at all: " + theirs.ours);
        }
    }

    /**
     * A clear which dropped nothing at all says why, where the only thing it had to say is a row it
     * could not act on: the catalog table went between the read of its rows and the loop that drops
     * what they named - an offline tool clearing the same backend, and the one state in which a clear
     * passes a row over and still drops nothing - so the account of it has no drop, no tree that had
     * lost its table and no leftover of this backend to be decided by. The row is what is left, and
     * the line reporting it is the only copy of what that row said.
     * <p>
     * What the case pins is one term of that condition. It pins it on a database holding nothing of
     * anybody else that the scan cannot attribute - the fragments asserted are the counts this case
     * owns, and a neighbour of another suite leaving an unstamped table would make the line fire for
     * a reason of its own rather than fail this.
     */
    @Test
    public void testAClearWhichDroppedNothingSaysWhyWhereARowItPassedOverIsAllItHad() throws Exception {
        final DN baseDN = DN.valueOf("dc=clear-catalog-race,dc=com");
        final TreeName owned = new TreeName(baseDN.toNormalizedUrlSafeString(), "id2entry");
        final ReportingStorage storage =
            new ReportingStorage(createBackendCfg(getBackendId() + "_catalogRace", baseDN));
        final String catalogTable = storage.getTableName(storage.getCatalogTree());
        final String ownedTable = storage.getTableName(owned);
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(owned, true);
                }
            });
            // the one row of the catalog now records a name outside the namespace this backend names
            // its tables in, so the read passes it over and the catalog names itself alone
            recordAnotherTable(catalogTable, "a_table_of_something_else");
            // and nothing of this backend is left standing for the scan to attribute to it
            dropTableBehindTheBackend(ownedTable);
            storage.close();

            // the table goes while the clear is running, which is what leaves the clear with nothing
            // dropped: an offline tool clearing the same backend a moment earlier. At the second
            // lookup and not the first, so that the rows are read before the table goes - the first
            // is catalogTables() asking whether there is a catalog at all
            storage.takeAwayAtLookupNumber(catalogTable, 2);

            storage.removeStorageFiles();

            assertFalse(isExistsTable(catalogTable), "the catalog table this case takes away was still there");
            storage.assertReported("a clear which dropped nothing at all and passed a row over said nothing"
                    + " about why, which is the silence of #888",
                "the clear removed no tree of this backend", "it dropped 0 table(s) in all",
                "0 of the trees its catalog names had lost their table already",
                "and 0 table(s) of this backend were named by no catalog");
            storage.assertReported("the row the clear could not act on was named by no line of it",
                "a_table_of_something_else", "passed over");
        } finally {
            clearQuietly(storage);
            dropTableIfExists(ownedTable);
            dropTableIfExists(catalogTable);
        }
    }

    /**
     * Takes every row out of a catalog, leaving the tables it named standing: what a backend upgraded
     * from a version keeping no catalog holds before its first read-write open fills one in.
     */
    private void emptyTheCatalog(String catalogTable) throws SQLException {
        try (final Connection con = DriverManager.getConnection(getJdbcUrl());
             final Statement st = con.createStatement()) {
            st.executeUpdate("delete from " + catalogTable);
        }
    }

    /**
     * Records the given table for every row of a catalog, as a version naming its tables otherwise
     * would have left them: the row names the right tree and a table this version never creates.
     */
    private void recordAnotherTable(String catalogTable, String tableName) throws SQLException {
        try (final Connection con = DriverManager.getConnection(getJdbcUrl());
             final PreparedStatement statement = con.prepareStatement("update " + catalogTable + " set v=?")) {
            statement.setBytes(1, tableName.getBytes(StandardCharsets.UTF_8));
            statement.executeUpdate();
        }
    }

    /** Empties the recorded table name of every row of a catalog, as a version recording none would have left it. */
    private void emptyTheRecordedTableNames(String catalogTable) throws SQLException {
        try (final Connection con = DriverManager.getConnection(getJdbcUrl());
             final PreparedStatement statement = con.prepareStatement("update " + catalogTable + " set v=?")) {
            statement.setBytes(1, new byte[0]);
            statement.executeUpdate();
        }
    }

    /**
     * Drops a table a clear left standing on purpose, so that it is not left behind for the rest of
     * the class. A failure here is swallowed rather than replacing the failure of the case it cleans
     * up after: what it leaves is dropped by the dropStaleTrees() of the next run of the class.
     */
    private void dropTableIfExists(String tableName) {
        try {
            if (isExistsTable(tableName)) {
                dropTableBehindTheBackend(tableName);
            }
        } catch (SQLException ignored) {
        }
    }

    /**
     * A storage which keeps the lines every clear it runs reports, so that a case can hold that
     * account to what it says.
     * <p>
     * Those lines change no state whatsoever, so a case asserting on the database a clear leaves
     * behind passes just as well with all of them deleted - which is how this report came to be
     * changed in three rounds of review with nothing able to fail. Here rather than in the case that
     * needed it first, for the same reason: the next line of the report wants an assertion too, and a
     * helper per case is what got the report where it was. See {@link JDBCStorage#reportClearLine}.
     */
    protected static final class ReportingStorage extends JDBCStorage {
        private final List<String> lines = Collections.synchronizedList(new ArrayList<String>());

        private volatile String tableToTakeAway;
        private final AtomicInteger lookupsToLetPass = new AtomicInteger();

        ReportingStorage(JDBCBackendCfg cfg) {
            super(cfg, null);
        }

        @Override
        void reportClearLine(LocalizableMessage line) {
            lines.add(line.toString());
            super.reportClearLine(line); // and on to the log, which is where an operator meets it
        }

        /**
         * Takes the named table away just before the given lookup of it, counting from the next one,
         * so that the lookup answers as another process taking the table a moment earlier would have
         * made it answer. What it models is the one state a clear cannot be put into from outside: a
         * table going between the read of the catalog and the loop that drops what that read named.
         * <p>
         * Which lookup matters, and the count is not decoration: a clear asks about its catalog table
         * twice - once in {@code catalogTables()} to decide whether there is a catalog to read at all,
         * and once in the drop loop, per entry. Taken away before the first, the clear reads no row,
         * passes none over and reports nothing, which is a different case from this one.
         * <p>
         * Dropped on the very connection the lookup is made on, and not on one of the test's own: the
         * clear holds its read of the catalog table until it commits, so a {@code drop table} issued
         * from a second session would queue behind the transaction that is waiting for this call to
         * return. Inside that transaction the drop takes no lock it does not already hold, and it is
         * committed with the loop - or, on the two engines committing DDL as they go, at once.
         */
        void takeAwayAtLookupNumber(String tableName, int nth) {
            lookupsToLetPass.set(nth - 1);
            tableToTakeAway = tableName;
        }

        @Override
        boolean isExistsTable(Connection con, JDBCStorage.TableScope scope, String tableName) {
            final String taking = tableToTakeAway;
            if (taking != null && taking.equalsIgnoreCase(tableName)
                    && lookupsToLetPass.getAndDecrement() <= 0) {
                tableToTakeAway = null; // once: every later lookup is answered by the database alone
                try (final PreparedStatement statement = con.prepareStatement("drop table " + taking)) {
                    statement.execute();
                } catch (SQLException e) {
                    throw new IllegalStateException("the table this case takes away could not be dropped", e);
                }
            }
            return super.isExistsTable(con, scope, tableName);
        }

        /** Every line reported so far, in the order the clears that reported them ran. */
        List<String> reported() {
            synchronized (lines) {
                return new ArrayList<>(lines);
            }
        }

        /**
         * Fails unless one reported line holds every one of the fragments. By fragments and not by the
         * whole line: what a case is entitled to pin is the thing the line is about, and a report
         * asserted word for word is a report nobody may improve the wording of.
         */
        void assertReported(String whatWentUnsaid, String... fragments) {
            for (final String line : reported()) {
                boolean holdsAll = true;
                for (final String fragment : fragments) {
                    holdsAll &= line.contains(fragment);
                }
                if (holdsAll) {
                    return;
                }
            }
            fail(whatWentUnsaid + "; the clear reported: " + reported());
        }
    }
}