~launchpad-pqm/launchpad/devel

# Copyright 2009 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

__metaclass__ = type

__all__ = ['MultiTableCopy']

import logging
import re
import time

from zope.interface import implements

from canonical.database import postgresql
from canonical.database.sqlbase import (cursor, quote, quoteIdentifier)
from canonical.launchpad.interfaces.looptuner import ITunableLoop
from canonical.launchpad.utilities.looptuner import DBLoopTuner


class PouringLoop:
    """Loop body to pour data from holding tables back into source tables.

    Used by MultiTableCopy internally to tell DBLoopTuner what to do.
    """
    implements(ITunableLoop)

    def __init__(self, from_table, to_table, transaction_manager, logger,
        batch_pouring_callback=None):

        self.from_table = str(from_table)
        self.to_table = str(to_table)
        self.transaction_manager = transaction_manager
        self.logger = logger
        self.batch_pouring_callback = batch_pouring_callback
        self.cur = cursor()

        self.cur.execute("SELECT min(id), max(id) FROM %s" % from_table)
        self.lowest_id, self.highest_id = self.cur.fetchone()

        if self.lowest_id is None:
            # Table is empty.
            self.lowest_id = 1
            self.highest_id = 0

        self.logger.debug("Up to %d rows in holding table"
                         % (self.highest_id + 1 - self.lowest_id))

    def isDone(self):
        """See `ITunableLoop`."""
        return self.lowest_id is None or self.lowest_id > self.highest_id

    def __call__(self, batch_size):
        """See `ITunableLoop`.

        Loop body: pour rows with ids up to "next" over to to_table."""
        batch_size = int(batch_size)

        # Figure out what id lies exactly batch_size rows ahead.
        self.cur.execute("""
            SELECT id
            FROM %s
            WHERE id >= %s
            ORDER BY id
            OFFSET %s
            LIMIT 1
            """ % (self.from_table, quote(self.lowest_id), quote(batch_size)))
        end_id = self.cur.fetchone()

        if end_id is not None:
            next = end_id[0]
        else:
            next = self.highest_id

        next += 1

        self.prepareBatch(
            self.from_table, self.to_table, batch_size, self.lowest_id, next)

        self.logger.debug("pouring %s: %d rows (%d-%d)" % (
            self.from_table, batch_size, self.lowest_id, next))

        self.cur.execute("INSERT INTO %s (SELECT * FROM %s WHERE id < %d)"
                         % (self.to_table, self.from_table, next))

        self.cur.execute("DELETE FROM %s WHERE id < %d"
                         % (self.from_table, next))

        self.lowest_id = next
        self._commit()

    def _commit(self):
        """Commit ongoing transaction, start a new one."""
        self.transaction_manager.commit()
        self.transaction_manager.begin()
        self.cur = cursor()
        # Disable slow sequential scans.  The database server is reluctant to
        # use indexes on tables that undergo large changes, such as the
        # deletion of large numbers of rows in this case.  Usually it's
        # right but in this case it seems to slow things down dramatically and
        # unnecessarily.  We disable sequential scans for every commit since
        # initZopeless by default resets our database connection with every
        # new transaction.
        # MultiTableCopy disables sequential scans for the first batch; this
        # just renews our setting after the connection is reset.
        postgresql.allow_sequential_scans(self.cur, False)

    def prepareBatch(
        self, from_table, to_table, batch_size, begin_id, end_id):
        """If batch_pouring_callback is defined, call it."""
        if self.batch_pouring_callback is not None:
            self.batch_pouring_callback(
                from_table, to_table, batch_size, begin_id, end_id)


class MultiTableCopy:
    """Copy interlinked data spanning multiple tables in a coherent fashion.

    This allows data from a combination of tables, possibly with foreign-key
    references between them, to be copied to a set of corresponding "holding
    tables"; processed and modified there; and then be inserted back to the
    original tables.  The holding tables are created on demand and dropped
    upon completion.

    You can tell the algorithm to redirect foreign keys.  Say you're copying a
    row x1 in a table X, and x1 has a foreign key referring to a row y1 in a
    table Y that you're also copying.  You will get copied rows x2 and y2
    respectively.  But if you declare the foreign-key relationship between X
    and Y to the algorithm, then x2's instance of that foreign key will refer
    not to y1 but to the new y2.  Any rows in X whose associated rows of Y are
    not copied, are also not copied.  This can be useful when copying data in
    entire sub-trees of the schema graph, e.g. "one distroseries and all the
    translations associated with it."

    All this happens in a two-stage process:

    1. Extraction stage.  Use the extract() method to copy selected data to a
    holding table, one table at a time.  Ordering matters: always do this in
    such an order that the table you are extracting has no foreign-key
    references to another table that you are yet to extract.

    This stage is relatively fast and holds no locks on the database.  Do any
    additional processing on the copied rows in the holding tables, during or
    after the extraction stage, so you do not hold any locks on the source
    tables yourself.  It's up to you to make sure that all of the rows in the
    holding tables can be inserted into their source tables: if you leave
    primary keys and such unchanged, unique constraints will be violated in
    the next stage.

    2. Pouring stage.  All data from the holding tables is inserted back into
    the source tables.  This entire stage, which normally takes the bulk of
    the copying time, is performed by calling the pour method.

    This stage will lock the rows that are being inserted in the source
    tables, if the database is so inclined (e.g. when using postgres with
    SERIALIZABLE isolation level).  For that reason, the pouring is done in
    smaller, controlled batches.  If you give the object a database
    transaction to work with, that transaction will be committed and restarted
    between batches.

    A MultiTableCopy is restartable.  If the process should fail for any
    reason, the holding tables will be left in one of two states: if stage 1
    has not completed, needsRecovery will return False.  In that case, drop
    the holding tables using dropHoldingTables and either start again (or give
    up).  But if a previous run did complete the extraction stage, the holding
    tables will remain and contain valid data.  In that case, just pour again
    to continue the work (and hopefully complete it this time).

    Holding tables will have names like "temp_POMsgSet_holding_ubuntu_feisty",
    in this case for one holding data extracted from source table POMsgSet by
    a MultiTableCopy called "ubuntu_feisty".  The holding tables are dropped
    when we're done, but they are not temp tables.  They have to be persistent
    so we get a chance to resume interrupted copies and analyse failures.

        # We use a regular, persistent table rather than a temp table for
        # this so that we get a chance to resume interrupted copies, and
        # analyse failures.  If we used temp tables, they'd be gone by the
        # time we knew something went wrong.

    The tables to be copied must meet a number of conventions:

     * First column must be an integer primary key called "id."

     * id values must be assigned by a sequence, with a name that can be used
       in SQL without quoting.

     * Every foreign-key column that refers to a table that is also being
       copied, has the same name as the table it refers to.  This can be
       changed by subclassing and overriding the _pointsToTable method.

     * Foreign-key column names and the tables they refer to can be used in
       SQL without quoting.

     * For any foreign key column "x" referring to another table that is also
       being copied, there must not be a column called "new_x"

    If you use this class, it is up to you to ensure that the data that you
    copy does not break compliance between the time it is extracted from its
    source tables and the moment it has been poured back there.  This means
    that rows that the data refers to by foreign keys must not be deleted
    while the multi-table copy is running, for instance.
    """

    def __init__(self, name, tables, seconds_per_batch=2.0,
            minimum_batch_size=500, restartable=True, logger=None):
        """Define a MultiTableCopy, including an in-order list of tables.

        :param name: a unique identifier for this MultiTableCopy operation,
            e.g. "ubuntu_feisty".  The name will be included in the names of
            holding tables.
        :param tables: a list of tables that will be extracted and poured,
            in the order in which they will be extracted (and later, poured).
            This is essential when analyzing recoverable state.  You may
            attempt multiple extractions from the same table, but all tables
            listed must be extracted.  If you do not wish to copy any rows
            from a source table, extract with "false" as its where clause.
        :param seconds_per_batch: a time goal (in seconds) to define how long,
            ideally, the algorithm should be allowed to hold locks on the
            source tables.  It will try to do the work in batches that take
            about this long.
        :param minimum_batch_size: minimum number of rows to pour in one
            batch.  You may want to set this to stop the algorithm from
            resorting to endless single-row batches in situations where
            response times are too slow but batch size turns out not to matter
            much.
        :param restartable: whether you want the remaining data to be
            available for recovery if the connection (or the process) fails
            while in the pouring stage.  If False, will extract to temp
            tables.  CAUTION: our connections currently get reset every time
            we commit a transaction, obliterating any temp tables possibly in
            the middle of the pouring process!
        :param logger: a logger to write informational output to.  If none is
            given, the default is used.
        """
        self.name = str(name)
        self.tables = tables
        self.lower_tables = [table.lower() for table in self.tables]
        self.seconds_per_batch = seconds_per_batch
        self.minimum_batch_size = minimum_batch_size
        self.last_extracted_table = None
        self.restartable = restartable
        self.batch_pouring_callbacks = {}
        self.pre_pouring_callbacks = {}
        if logger is not None:
            self.logger = logger
        else:
            self.logger = logging

    def dropHoldingTables(self):
        """Drop any holding tables that may exist for this MultiTableCopy."""
        holding_tables = [self.getHoldingTableName(table)
                          for table in self.tables]
        postgresql.drop_tables(cursor(), holding_tables)

    def getRawHoldingTableName(self, tablename, suffix=''):
        """Name for a holding table, but without quotes.  Use with care."""
        if suffix:
            suffix = '_%s' % suffix

        assert re.search(r'[^a-z_]', tablename + suffix) is None, (
            'Unsupported characters in table name per Bug #179821')

        raw_name = "temp_%s_holding_%s%s" % (
            str(tablename), self.name, str(suffix))

        return raw_name

    def getHoldingTableName(self, tablename, suffix=''):
        """Name for a holding table to hold data being copied in tablename.

        Return value is properly quoted for use as an SQL identifier.
        """
        raw_name = self.getRawHoldingTableName(tablename, suffix)
        return quoteIdentifier(raw_name)

    def _pointsToTable(self, source_table, foreign_key):
        """Name of table that source_table.foreign_key refers to.

        By default, all foreign keys that play a role in the MultiTableCopy
        are expected to have the same names as the tables they refer to.  If
        that is not the case for your copy, subclass this class and override
        this method with a version that has specific knowledge of what points
        where.
        """
        return foreign_key

    def extract(self, source_table, joins=None, where_clause=None,
        id_sequence=None, inert_where=None, pre_pouring_callback=None,
        batch_pouring_callback=None, external_joins=None):
        """Extract (selected) rows from source_table into a holding table.

        The holding table gets an additional new_id column with identifiers
        generated by `id_sequence`.  Apart from this extra column, (and
        indexes and constraints), the holding table is schematically identical
        to `source_table`.  A unique index is created for the original id
        column.

        There is a special facility for redirecting foreign keys to other
        tables in the same copy operation.  The `joins` argument can pass a
        list of foreign keys.  The foreign keys given in joins must be columns
        of `source_table`, and refer to tables that are also being copied.
        The selection used in populating the holding table for `source_table`
        will be joined on each of the foreign keys listed in `joins`.  The
        foreign keys in the holding table will point to the new_ids of the
        copied rows, rather than the original ids.  Rows in `source_table`
        will only be copied to their holding table if all rows they are joined
        with through the `joins` parameter are also copied.

        When joining, the added tables' columns are not included in the
        holding table, but `where_clause` may still select on them.

        :param source_table: table to extract data from (and where the
            extracted data will ultimately be poured back to).
        :param joins: list of foreign keys from `source_table` to other source
            tables being copied in the same operation.  By default, each of
            these foreign keys is assumed to refer to a table of the same
            name.  When breaking this convention, override the
            `_pointsToTable` method to provide the right target table for each
            foreign key that the operation should know about.
        :param where_clause: Boolean SQL expression characterizing rows to be
            extracted.  The WHERE clause may refer to rows from table being
            extracted as "source."
        :param id_sequence: SQL sequence that should assign new identifiers
            for the extracted rows.  Defaults to `source_table` with "_seq_id"
            appended, which by SQLObject/Launchpad convention is the sequence
            that provides `source_table`'s primary key values.  Used verbatim,
            without quoting.
        :param inert_where: Boolean SQL expression characterizing rows that
            are extracted, but should not poured back into `source_table`
            during the pouring stage.  For these rows, new_id will be null.
            This clause is executed in a separate query, therefore will have
            no access to any tables other than the newly created holding
            table.  The clause can reference the holding table under the name
            "holding."  Any foreign keys from `joins` will still contain the
            values they had in `source_table`, but for each "x" of these
            foreign keys, the holding table will have a column "new_x" that
            holds the redirected foreign key.
        :param pre_pouring_callback: a callback that is called just before
            pouring this table.  At that time the holding table will no longer
            have its new_id column, its values having been copied to the
            regular id column.  This means that the copied rows' original ids
            are no longer known.  The callback takes as arguments the holding
            table's name and the source table's name.  The callback may be
            invoked more than once if pouring is interrupted and later
            resumed.
        :param batch_pouring_callback: a callback that is called before each
            batch of rows is poured, within the same transaction that pours
            those rows.  It takes as arguments the holding table's name; the
            source table's name; current batch size; the id of the first row
            being poured; and the id where the batch stops.  The "end id" is
            exclusive, so a row with that id is not copied (and in fact may
            not even exist).  The time spent by `batch_ouring_callback` is
            counted as part of the batch's processing time.
        :param external_joins: a list of tables to join into the extraction
            query, so that the extraction condition can select on them.  Each
            entry must be a string either simply naming a table ("Person"), or
            naming a table and providing a name for it in the query (e.g.,
            "Person p").  Do the latter if the same table also occurs
            elsewhere in the same query.  Your `where_clause` can refer to the
            rows included in this way by the names you give or by their table
            name, SQL rules permitting.  If your join yields multiple rows
            that have the same `source_table` id, only one arbitrary pick of
            those will be extracted.
        """
        if id_sequence is None:
            id_sequence = "%s_id_seq" % source_table.lower()

        if joins is None:
            joins = []

        self.batch_pouring_callbacks[source_table] = batch_pouring_callback
        self.pre_pouring_callbacks[source_table] = pre_pouring_callback

        if external_joins is None:
            external_joins = []

        source_table = str(source_table)
        self._checkExtractionOrder(source_table)

        holding_table = self.getHoldingTableName(source_table)

        self.logger.info('Extracting from %s into %s...' % (
            source_table, holding_table))

        starttime = time.time()

        cur = self._selectToHolding(source_table, joins, external_joins,
            where_clause, holding_table, id_sequence, inert_where)

        if len(joins) > 0:
            self._retargetForeignKeys(holding_table, joins, cur)

        # Now that our new holding table is in a stable state, index its id
        self._indexIdColumn(holding_table, source_table, cur)

        self.logger.debug(
            '...Extracted in %.3f seconds' % (time.time() - starttime))

    def _selectToHolding(self, source_table, joins, external_joins,
            where_clause, holding_table, id_sequence, inert_where):
        """Create holding table based on data from source table.

        We don't need to know what's in the source table exactly; we just
        create a new table straight from a "select *."  Except we also add a
        few columns for the purpose of retargeting foreign keys, as well as a
        new_id column.
        """
        source_table = str(source_table)
        select = ['source.*']
        from_list = ["%s source" % source_table]
        where = []

        for join in joins:
            join = str(join)
            referenced_table = self._pointsToTable(source_table, join)
            referenced_holding = self.getHoldingTableName(referenced_table)
            column = join.lower()

            self._checkForeignKeyOrder(column, referenced_table)

            select.append('%s.new_id AS new_%s' % (
                referenced_holding, column))
            from_list.append(referenced_holding)
            where.append('%s = %s.id' % (column, referenced_holding))

        from_list.extend(external_joins)

        if where_clause is not None:
            where.append('(%s)' % where_clause)

        where_text = ''
        if len(where) > 0:
            where_text = 'WHERE %s' % ' AND '.join(where)

        # For each row we append at the end any new foreign key values, and
        # finally a "new_id" holding its future id field.  This new_id value
        # is allocated from the original table's id sequence, so it will be
        # unique in the original table.
        table_creation_parameters = {
            'columns': ','.join(select),
            'holding_table': holding_table,
            'id_sequence': "nextval('%s'::regclass)" % id_sequence,
            'inert_where': inert_where,
            'source_tables': ','.join(from_list),
            'where': where_text,
            'temp': '',
            }
        if not self.restartable:
            table_creation_parameters['temp'] = 'TEMP'

        cur = cursor()

        # Create holding table directly from select result.
        if inert_where is None:
            # We'll be pouring all rows from this table.  To avoid a costly
            # second write pass (which would rewrite all records in the
            # holding table), we assign new_ids right in the same query.
            cur.execute('''
                CREATE %(temp)s TABLE %(holding_table)s AS
                SELECT DISTINCT ON (source.id)
                    %(columns)s, %(id_sequence)s AS new_id
                FROM %(source_tables)s
                %(where)s
                ORDER BY id''' % table_creation_parameters)
        else:
            # Some of the rows may have to have null new_ids.  To avoid
            # wasting "address space" on the sequence, we populate the entire
            # holding table with null new_ids, then fill in new_id only for
            # rows that do not match the "inert_where" condition.
            cur.execute('''
                CREATE %(temp)s TABLE %(holding_table)s AS
                SELECT DISTINCT ON (source.id)
                    %(columns)s, NULL::integer AS new_id
                FROM %(source_tables)s
                %(where)s
                ORDER BY id''' % table_creation_parameters)
            cur.execute('''
                UPDATE %(holding_table)s AS holding
                SET new_id = %(id_sequence)s
                WHERE NOT (%(inert_where)s)
                ''' % table_creation_parameters)

        return cur

    def _indexIdColumn(self, holding_table, source_table, cur):
        """Index id column on holding table.

        Creates a unique index on "id" column in holding table.  The index
        gets the name of the holding table with "_id" appended.
        """
        source_table = str(source_table)
        self.logger.debug("Indexing %s" % holding_table)
        cur.execute('''
            CREATE UNIQUE INDEX %s
            ON %s (id)
        ''' % (self.getHoldingTableName(source_table, 'id'), holding_table))

    def _retargetForeignKeys(self, holding_table, joins, cur):
        """Replace foreign keys in new holding table.

        Set the values of the foreign keys that are to be retargeted to those
        in their respective "new_" variants, then drop those "new_" columns we
        added from the holding table.
        """
        columns = [join.lower() for join in joins]
        fk_updates = ['%s = new_%s' % (column, column) for column in columns]
        updates = ', '.join(fk_updates)
        self.logger.debug("Redirecting foreign keys: %s" % updates)
        cur.execute("UPDATE %s SET %s" % (holding_table, updates))
        for column in columns:
            self.logger.debug("Dropping foreign-key column %s" % column)
            cur.execute("ALTER TABLE %s DROP COLUMN new_%s"
                        % (holding_table, column))

    def needsRecovery(self):
        """Do we have holding tables with recoverable data from previous run?

        Returns Boolean answer.
        """

        cur = cursor()

        # If there are any holding tables to be poured into their source
        # tables, there must at least be one for the last table that pour()
        # processes.
        last_holding_table = self.getRawHoldingTableName(self.tables[-1])
        if not postgresql.have_table(cur, last_holding_table):
            return False

        # If the first table in our list also still exists, and it still has
        # its new_id column, then the pouring process had not begun yet.
        # Assume the data was not ready for pouring.
        first_holding_table = self.getRawHoldingTableName(self.tables[0])
        if postgresql.table_has_column(cur, first_holding_table, 'new_id'):
            self.logger.info(
                "Previous run aborted too early for recovery; redo all")
            return False

        self.logger.info("Recoverable data found")
        return True

    def pour(self, transaction_manager):
        """Pour data from holding tables back into source tables.

        Rows in the holding table that have their new_id set to null are
        skipped.

        The transaction manager is committed and re-opened after every batch
        run.

        Batch sizes are dynamically adjusted to meet the stated time goal.
        """
        if self.last_extracted_table is None:
            if not self.needsRecovery():
                raise AssertionError("Can't pour: no tables extracted")
        elif self.last_extracted_table != len(self.tables) - 1:
            raise AssertionError(
                "Not safe to pour: last table '%s' was not extracted"
                % self.tables[-1])

        cur = self._commit(transaction_manager)

        # Don't let postgres revert to slow sequential scans while we pour.
        # That might otherwise happen to the holding table as its vital "id"
        # index degrades with the removal of rows.
        postgresql.allow_sequential_scans(cur, False)

        # Main loop: for each of the source tables being copied, see if
        # there's a matching holding table.  If so, prepare it, pour it back
        # into the source table, and drop.
        for table in self.tables:
            holding_table_unquoted = self.getRawHoldingTableName(table)

            if not postgresql.have_table(cur, holding_table_unquoted):
                # We know we're in a suitable state for pouring.  If this
                # table does not exist, it must be because it's been poured
                # out completely and dropped in an earlier instance of this
                # loop, before the failure we're apparently recovering from.
                continue

            holding_table = self.getHoldingTableName(table)
            self.logger.info("Pouring %s back into %s..."
                         % (holding_table, table))

            tablestarttime = time.time()

            has_new_id = postgresql.table_has_column(
                cur, holding_table_unquoted, 'new_id')

            self._pourTable(
                holding_table, table, has_new_id, transaction_manager)

            # Drop holding table.  It may still contain rows with id set to
            # null.  Those must not be poured.
            postgresql.drop_tables(cursor(), holding_table)

            self.logger.debug(
                "Pouring %s took %.3f seconds."
                % (holding_table, time.time() - tablestarttime))

            cur = self._commit(transaction_manager)

        # In future, let the database perform sequential scans again if it
        # decides that's best.
        postgresql.allow_sequential_scans(cur, True)

    def _pourTable(
        self, holding_table, table, has_new_id, transaction_manager):
        """Pour contents of a holding table back into its source table.

        This will commit transaction_manager, typically multiple times.
        """
        if has_new_id:
            # Update ids in holding table from originals to copies.  To
            # broaden the caller's opportunity to manipulate rows in the
            # holding tables, we skip rows that have new_id set to null.
            cur = cursor()
            cur.execute("UPDATE %s SET id=new_id" % holding_table)
            # Restore table to original schema
            cur.execute("ALTER TABLE %s DROP COLUMN new_id" % holding_table)
            self._commit(transaction_manager)
            self.logger.debug("...rearranged ids...")

        callback = self.pre_pouring_callbacks.get(table)
        if callback is not None:
            callback(holding_table, table)

        # Now pour holding table's data into its source table.  This is where
        # we start writing to tables that other clients will be reading, and
        # our transaction will usually be serializable, so row locks are a
        # concern.  Break the writes up in batches of at least a thousand
        # rows.  The goal is to have these transactions running no longer than
        # five seconds or so each; we aim for four just to be sure.

        pourer = PouringLoop(
            holding_table, table, transaction_manager, self.logger,
            self.batch_pouring_callbacks.get(table))
        DBLoopTuner(
            pourer, self.seconds_per_batch, self.minimum_batch_size).run()

    def _checkExtractionOrder(self, source_table):
        """Verify order in which tables are extracted against tables list.

        Check that the caller follows the stated plan, extracting from tables
        in the same order as in self.tables.
        """
        try:
            table_number = self.tables.index(source_table)
        except ValueError:
            raise AssertionError(
                "Can't extract '%s': not in list of tables" % source_table)

        if self.last_extracted_table is None:
            # Can't skip the first table!
            if table_number > 0:
                raise AssertionError(
                    "Can't extract: skipped first table '%s'"
                    % self.tables[0])
        else:
            if table_number < self.last_extracted_table:
                raise AssertionError(
                    "Table '%s' extracted after its turn" % source_table)
            if table_number > self.last_extracted_table + 1:
                raise AssertionError(
                    "Table '%s' extracted before its turn" % source_table)
            if table_number == self.last_extracted_table:
                raise AssertionError(
                    "Table '%s' extracted again" % source_table)

        self.last_extracted_table = table_number

    def _checkForeignKeyOrder(self, fk, referenced_table):
        """Verify that we're in a position to "retarget" a foreign key.

        We've been asked to retarget a foreign key while copying.  Check that
        the table it refers to has already been copied.
        """
        lower_name = referenced_table.lower()
        lower_tables = self.lower_tables
        try:
            target_number = lower_tables.index(lower_name)
        except ValueError:
            raise AssertionError(
                "Foreign key '%s' refers to table '%s' "
                "which is not being copied" % (fk, referenced_table))

        if target_number > self.last_extracted_table:
            raise AssertionError(
                "Foreign key '%s' refers to table '%s' "
                "which is to be copied later" % (fk, referenced_table))
        if target_number == self.last_extracted_table:
            raise AssertionError(
                "Foreign key '%s' in table '%s' "
                "is a self-reference" % (fk, referenced_table))

    def _commit(self, transaction_manager):
        """Commit our transaction and create replacement cursor.

        Use this as "cur = self._commit(transaction_manager)" to commit a
        transaction, restart it, and create a cursor that lives within the new
        transaction.
        """
        start = time.time()
        transaction_manager.commit()
        self.logger.debug("Committed in %.3f seconds" % (time.time() - start))
        transaction_manager.begin()
        return cursor()