File Coverage

blib/lib/DBIx/BatchChunker.pm

Criterion	Covered	Total	%
statement	349	372	93.8
branch	156	192	81.2
condition	52	79	65.8
subroutine	45	47	95.7
pod	3	4	75.0
total	605	694	87.1

package DBIx::BatchChunker;

our $AUTHORITY = 'cpan:GSG';
# ABSTRACT: Run large database changes safely
use version;
our $VERSION = 'v1.0.2'; # VERSION

use Moo;
use MooX::StrictConstructor;

use CLDR::Number;

use Types::Standard        qw( Str Bool Undef ArrayRef HashRef CodeRef InstanceOf Tuple Maybe Optional slurpy );
use Types::Numbers         qw( NumRange UnsignedInt PerlSafeInt PositiveInt PositiveOrZeroNum );
use Type::Utils;

use List::Util        1.33 (qw( min max sum any first ));  # has any/all/etc.
use Math::BigInt upgrade => 'Math::BigFloat';
use Math::BigFloat;
use POSIX                   qw( ceil );
use Scalar::Util            qw( blessed weaken );
use Term::ProgressBar 2.14;                          # with silent option
use Time::HiRes             qw( time sleep );

use DBIx::BatchChunker::LoopState;

# Don't export the above, but don't conflict with StrictConstructor, either
use namespace::clean -except => [qw< new meta >];

# This is now an unused, dummy variable
our $DB_MAX_ID = ~0;

#pod =encoding utf8
#pod
#pod =head1 SYNOPSIS
#pod
#pod     use DBIx::BatchChunker;
#pod
#pod     my $account_rs = $schema->resultset('Account')->search({
#pod         account_type => 'deprecated',
#pod     });
#pod
#pod     my %params = (
#pod         chunk_size  => 5000,
#pod         target_time => 5,
#pod
#pod         rs      => $account_rs,
#pod         id_name => 'account_id',
#pod
#pod         coderef => sub { $_[1]->delete },
#pod         sleep   => 1,
#pod         verbose => 1,
#pod
#pod         progress_name    => 'Deleting deprecated accounts',
#pod         process_past_max => 1,
#pod     );
#pod
#pod     # EITHER:
#pod     # 1) Automatically construct and execute the changes:
#pod
#pod     DBIx::BatchChunker->construct_and_execute(%params);
#pod
#pod     # OR
#pod     # 2) Manually construct and execute the changes:
#pod
#pod     my $batch_chunker = DBIx::BatchChunker->new(%params);
#pod
#pod     $batch_chunker->calculate_ranges;
#pod     $batch_chunker->execute;
#pod
#pod =head1 DESCRIPTION
#pod
#pod This utility class is for running a large batch of DB changes in a manner that doesn't
#pod cause huge locks, outages, and missed transactions.  It's highly flexible to allow for
#pod many different kinds of change operations, and dynamically adjusts chunks to its
#pod workload.
#pod
#pod It works by splitting up DB operations into smaller chunks within a loop.  These chunks
#pod are transactionalized, either naturally as single-operation bulk work or by the loop
#pod itself.  The full range is calculated beforehand to get the right start/end points.
#pod A L will be created to let the deployer know the
#pod processing status.
#pod
#pod There are two ways to use this class: call the automatic constructor and executor
#pod (L) or manually construct the object and call its methods. See
#pod L for examples of both.
#pod
#pod B You should not rely on this class to magically fix any and all locking
#pod problems the DB might experience just because it's being used.  Thorough testing and
#pod best practices are still required.
#pod
#pod =head2 Processing Modes
#pod
#pod This class has several different modes of operation, depending on what was passed to
#pod the constructor:
#pod
#pod =head3 DBIC Processing
#pod
#pod If both L and L are passed, a chunk ResultSet is built from the base
#pod ResultSet, to add in a C clause, and the new ResultSet is passed into the
#pod coderef.  The coderef should run some sort of active ResultSet operation from there.
#pod
#pod An L should be provided, but if it is missing it will be looked up based on
#pod the primary key of the ResultSource.
#pod
#pod If L is also enabled, then each chunk is wrapped in a transaction and the
#pod coderef is called for each row in the chunk.  In this case, the coderef is passed a
#pod Result object instead of the chunk ResultSet.
#pod
#pod Note that whether L is enabled or not, the coderef execution is encapsulated
#pod in DBIC's retry logic, so any failures will re-connect and retry the coderef.  Because of
#pod this, any changes you make within the coderef should be idempotent, or should at least be
#pod able to skip over any already-processed rows.
#pod
#pod =head3 Active DBI Processing
#pod
#pod If an L (DBI statement handle args) is passed without a L, the statement
#pod handle is merely executed on each iteration with the start and end IDs.  It is assumed
#pod that the SQL for the statement handle contains exactly two placeholders for a C
#pod clause.  For example:
#pod
#pod     my $update_stmt = q{
#pod     UPDATE
#pod         accounts a
#pod         JOIN account_updates au USING (account_id)
#pod     SET
#pod         a.time_stamp = au.time_stamp
#pod     WHERE
#pod         a.account_id BETWEEN ? AND ? AND
#pod         a.time_stamp != au.time_stamp
#pod     });
#pod
#pod The C clause should, of course, match the IDs being used in the loop.
#pod
#pod The statement is ran with L for retry protection.  Therefore, the
#pod statement should also be idempotent.
#pod
#pod =head3 Query DBI Processing
#pod
#pod If both a L and a L are passed, the statement handle is prepared and
#pod executed.  Like the L mode, the SQL for the statement should
#pod contain exactly two placeholders for a C clause.  Then the C<$sth> is passed to
#pod the coderef.  It's up to the coderef to extract data from the executed statement handle,
#pod and do something with it.
#pod
#pod If C is enabled, each chunk is wrapped in a transaction and the coderef is
#pod called for each row in the chunk.  In this case, the coderef is passed a hashref of the
#pod row instead of the executed C<$sth>, with lowercase alias names used as keys.
#pod
#pod Note that in both cases, the coderef execution is encapsulated in a L
#pod call to either C or C (using L), so any failures will
#pod re-connect and retry the coderef.  Because of this, any changes you make within the
#pod coderef should be idempotent, or should at least be able to skip over any
#pod already-processed rows.
#pod
#pod =head3 DIY Processing
#pod
#pod If a L is passed but neither a C nor a C are passed, then the
#pod multiplier loop does not touch the database.  The coderef is merely passed the start and
#pod end IDs for each chunk.  It is expected that the coderef will run through all database
#pod operations using those start and end points.
#pod
#pod It's still valid to include L, L, and/or L in the
#pod constructor to enable features like L or
#pod L.
#pod
#pod If you're not going to include any min/max statements for L, you will
#pod need to set L and L yourself, either in the constructor or before the
#pod L call.  Using L is also not an option in this case, as
#pod this tries to call L without a way to do so.
#pod
#pod =head3 TL;DR Version
#pod
#pod     $stmt                             = Active DBI Processing
#pod     $stmt + $coderef                  = Query DBI Processing  | $bc->$coderef($executed_sth)
#pod     $stmt + $coderef + single_rows=>1 = Query DBI Processing  | $bc->$coderef($row_hashref)
#pod     $rs   + $coderef                  = DBIC Processing       | $bc->$coderef($chunk_rs)
#pod     $rs   + $coderef + single_rows=>1 = DBIC Processing       | $bc->$coderef($result)
#pod             $coderef                  = DIY Processing        | $bc->$coderef($start, $end)
#pod
#pod =head1 ATTRIBUTES
#pod
#pod See the L section for more in-depth descriptions of these attributes and their
#pod usage.
#pod
#pod =head2 DBIC Processing Attributes
#pod
#pod =head3 rs
#pod
#pod A L.  This is used by all methods as the base ResultSet onto which
#pod the DB changes will be applied.  Required for DBIC processing.
#pod
#pod =cut

has rs => (
    is       => 'ro',
    isa      => InstanceOf['DBIx::Class::ResultSet'],
    required => 0,
);

#pod =head3 rsc
#pod
#pod A L.  This is only used to override L for min/max
#pod calculations.  Optional.
#pod
#pod =cut

has rsc => (
    is        => 'ro',
    isa       => InstanceOf['DBIx::Class::ResultSetColumn'],
    required  => 0,
);

#pod =head3 count_rs
#pod
#pod A L, only used to override L for row counting calculations.
#pod For 99.9% of cases, you do not need to set this.  Though, it could be used for the rare
#pod case where the original Resulset would run into indexing problems with its row counting
#pod statement and needs something broader to compensate.
#pod
#pod B Do not set this unless you know what you're doing.  Having a different C
#pod ResultSet from the base ResultSet means that the row counts to size up the chunk workload
#pod will be different from the workload itself.  If the row counts are too high, you may end
#pod up with workloads that are too quick and L may compensate
#pod with an overly large chunk size, anyway.  If the row counts are too low, you risk having
#pod a oversized chunk that gets processed and locks rows for too long, and chunk resizing may
#pod even skip blocks that it thinks have no rows to process.
#pod
#pod =cut

has count_rs => (
    is       => 'ro',
    isa      => InstanceOf['DBIx::Class::ResultSet'],
    required => 0,
);

#pod =head3 dbic_retry_opts
#pod
#pod A hashref of DBIC retry options.  These options control how retry protection works within
#pod DBIC.  So far, there are two supported options:
#pod
#pod     max_attempts  = Number of times to retry
#pod     retry_handler = Coderef that returns true to continue to retry or false to re-throw
#pod                     the last exception
#pod
#pod The default is to let the DBIC storage engine handle its own protection, which will retry
#pod once if the DB connection was disconnected.  If you specify any options, even a blank
#pod hashref, BatchChunker will fill in a default C of 10, and an always-true
#pod C.  This is similar to L's defaults.
#pod
#pod Under the hood, these are options that are passed to the as-yet-undocumented
#pod L.  The C has access to the same
#pod BlockRunner object (passed as its only argument) and its methods/accessors, such as C,
#pod C, and C.
#pod
#pod =cut

has dbic_retry_opts => (
    is        => 'ro',
    isa       => HashRef,
    required  => 0,
    predicate => '_has_dbic_retry_opts',
);

sub _dbic_block_runner {
    my ($self, $method, $coderef) = @_;

    my $storage = $self->dbic_storage;

    # Block running disabled
    unless ($self->_has_dbic_retry_opts) {
        return $storage->txn_do($coderef) if $method eq 'txn';
        return $storage->dbh_do($coderef);
    }

    # A very light wrapper around BlockRunner.  No need to load BlockRunner, since DBIC
    # loads it in before us if we're using this method.
    DBIx::Class::Storage::BlockRunner->new(
        # in case they are not defined with a custom dbic_retry_opts
        max_attempts  => 10,
        retry_handler => sub { 1 },

        # never overrides the important ones below
        %{ $self->dbic_retry_opts },

        storage  => $storage,
        wrap_txn => ($method eq 'txn' ? 1 : 0),
    )->run($coderef);
}

#pod =head2 DBI Processing Attributes
#pod
#pod =head3 dbi_connector
#pod
#pod A L object.  Instead of L statement handles, this is the
#pod recommended way for BatchChunker to interface with the DBI, as it handles retries on
#pod failures.  The connection mode used is whatever default is set within the object.
#pod
#pod Required for DBI Processing, unless L is specified.
#pod
#pod =cut

has dbi_connector => (
    is       => 'ro',
    isa      => InstanceOf['DBIx::Connector::Retry'],
    required => 0,
);

#pod =head3 dbic_storage
#pod
#pod A DBIC storage object, as an alternative for L.  There may be times when
#pod you want to run plain DBI statements, but are still using DBIC.  In these cases, you
#pod don't have to create a L object to run those statements.
#pod
#pod This uses a BlockRunner object for retry protection, so the options in
#pod L would apply here.
#pod
#pod Required for DBI Processing, unless L is specified.
#pod
#pod =cut

has dbic_storage => (
    is       => 'ro',
    isa      => InstanceOf['DBIx::Class::Storage::DBI'],
    required => 0,
);

#pod =head3 min_stmt
#pod
#pod =head3 max_stmt
#pod
#pod SQL statement strings or an arrayref of parameters for L.
#pod
#pod When executed, these statements should each return a single value, either the minimum or
#pod maximum ID that will be affected by the DB changes.  These are used by
#pod L.  Required if using either type of DBI Processing.
#pod
#pod =cut

my $SQLStringOrSTHArgs_type = Type::Utils::declare(
    name       => 'SQLStringOrSTHArgs',
    # Allow an SQL string, an optional hashref/undef, and any number of strings/undefs
    parent     => Tuple->parameterize(Str, Optional[Maybe[HashRef]], slurpy ArrayRef[Maybe[Str]]),
    coercion   => sub { $_ = [ $_ ] if Str->check($_); $_ },
    message    => sub { 'Must be either an SQL string or an arrayref of parameters for $sth creation (SQL + hashref/undef + binds)' },
);

has min_stmt => (
    is       => 'ro',
    isa      => $SQLStringOrSTHArgs_type,
    required => 0,
    coerce   => 1,
);

has max_stmt => (
    is       => 'ro',
    isa      => $SQLStringOrSTHArgs_type,
    required => 0,
    coerce   => 1,
);

#pod =head3 stmt
#pod
#pod A SQL statement string or an arrayref of parameters for L + binds.
#pod
#pod If using L (no coderef), this is a L statement
#pod (usually DML like C).  If using L (with
#pod coderef), this is a passive DQL (C) statement.
#pod
#pod In either case, the statement should contain C placeholders, which will be
#pod executed with the start/end ID points.  If there are already bind placeholders in the
#pod arrayref, then make sure the C bind points are last on the list.
#pod
#pod Required for DBI Processing.
#pod
#pod =cut

has stmt => (
    is       => 'ro',
    isa      => $SQLStringOrSTHArgs_type,
    required => 0,
    coerce   => 1,
);

#pod =head3 count_stmt
#pod
#pod A C SQL statement string or an arrayref of parameters for
#pod L.
#pod
#pod Like L, it should contain C placeholders.  In fact, the SQL should look
#pod exactly like the L query, except with C instead of the column list.
#pod
#pod Used only for L.  Optional, but recommended for
#pod L.
#pod
#pod =cut

has count_stmt => (
    is       => 'ro',
    isa      => $SQLStringOrSTHArgs_type,
    required => 0,
    coerce   => 1,
);

#pod =head2 Progress Bar Attributes
#pod
#pod =head3 progress_bar
#pod
#pod The progress bar used for all methods.  This can be specified right before the method
#pod call to override the default used for that method.  Unlike most attributes, this one
#pod is read-write, so it can be switched on-the-fly.
#pod
#pod Don't forget to remove or switch to a different progress bar if you want to use a
#pod different one for another method:
#pod
#pod     $batch_chunker->progress_bar( $calc_pb );
#pod     $batch_chunker->calculate_ranges;
#pod     $batch_chunker->progress_bar( $loop_pb );
#pod     $batch_chunker->execute;
#pod
#pod All of this is optional.  If the progress bar isn't specified, the method will create
#pod a default one.  If the terminal isn't interactive, the default L will
#pod be set to C to naturally skip the output.
#pod
#pod =cut

has progress_bar => (
    is       => 'rw',
    isa      => InstanceOf['Term::ProgressBar'],
);

#pod =head3 progress_name
#pod
#pod A string used by L to assist in creating a progress bar.  Ignored if
#pod L is already specified.
#pod
#pod This is the preferred way of customizing the progress bar without having to create one
#pod from scratch.
#pod
#pod =cut

has progress_name => (
    is       => 'rw',
    isa      => Str,
    required => 0,
    lazy     => 1,
    default  => sub {
        my $rs = shift->rs;
        'Processing'.(defined $rs ? ' '.$rs->result_source->name : '');
    },
);

#pod =head3 cldr
#pod
#pod A L object.  English speakers that use a typical C<1,234.56> format would
#pod probably want to leave it at the default.  Otherwise, you should provide your own.
#pod
#pod =cut

has cldr => (
    is       => 'rw',
    isa      => InstanceOf['CLDR::Number'],
    required => 0,
    lazy     => 1,
    default  => sub { CLDR::Number->new(locale => 'en') },
);

#pod =head3 verbose
#pod
#pod Boolean.  By default, this is on, which displays timing stats on each chunk, as well as
#pod total numbers.  This is still subject to non-interactivity checks from L.
#pod
#pod (This was previously defaulted to off, and called C, prior to v1.0.0.)
#pod
#pod =for Pod::Coverage debug
#pod
#pod =cut

has verbose => (
    is       => 'rw',
    isa      => Bool,
    required => 0,
    default  => 1,
);

# Backwards-compatibility
*debug = \&verbose;

#pod =head2 Common Attributes
#pod
#pod =head3 id_name
#pod
#pod The column name used as the iterator in the processing loops.  This should be a primary
#pod key or integer-based (indexed) key, tied to the L.
#pod
#pod Optional.  Used mainly in DBIC processing.  If not specified, it will look up
#pod the first primary key column from L and use that.
#pod
#pod This can still be specified for other processing modes to use in progress bars.
#pod
#pod =cut

has id_name => (
    is       => 'rw',
    isa      => Str,
    required => 0,
    trigger  => \&_fix_id_name,
);

sub _fix_id_name {
    my ($self, $id_name) = @_;
    return if !$id_name || $id_name =~ /\./ || !defined $self->rs;  # prevent an infinite trigger loop
    $self->id_name( $self->rs->current_source_alias.".$id_name" );
}

#pod =head3 coderef
#pod
#pod The coderef that will be called either on each chunk or each row, depending on how
#pod L is set.  The first input is always the BatchChunker object.  The rest
#pod vary depending on the processing mode:
#pod
#pod     $stmt + $coderef                  = Query DBI Processing  | $bc->$coderef($executed_sth)
#pod     $stmt + $coderef + single_rows=>1 = Query DBI Processing  | $bc->$coderef($row_hashref)
#pod     $rs   + $coderef                  = DBIC Processing       | $bc->$coderef($chunk_rs)
#pod     $rs   + $coderef + single_rows=>1 = DBIC Processing       | $bc->$coderef($result)
#pod             $coderef                  = DIY Processing        | $bc->$coderef($start, $end)
#pod
#pod The loop does not monitor the return values from the coderef.
#pod
#pod Required for all processing modes except L.
#pod
#pod =cut

has coderef => (
    is       => 'ro',
    isa      => CodeRef,
    required => 0,
);

#pod =head3 chunk_size
#pod
#pod The amount of rows to be processed in each loop.
#pod
#pod This figure should be sized to keep per-chunk processing time at around 5 seconds.  If
#pod this is too large, rows may lock for too long.  If it's too small, processing may be
#pod unnecessarily slow.
#pod
#pod Default is 1 row, which is only appropriate if L (on by default) is
#pod enabled.  This will cause the processing to slowly ramp up to the target time as
#pod BatchChunker gathers more data.
#pod
#pod Otherwise, if you using static chunk sizes with C turned off, figure out
#pod the right chunk size with a few test runs and set it here.
#pod
#pod (This was previously defaulted to 1000 rows, prior to v1.0.0.)
#pod
#pod =cut

has chunk_size => (
    is       => 'rw',
    isa      => PositiveInt,
    required => 0,
    default  => 1,
);

#pod =head3 target_time
#pod
#pod The target runtime (in seconds) that chunk processing should strive to achieve, not
#pod including L.  If the chunk processing times are too high or too low, this will
#pod dynamically adjust L to try to match the target.
#pod
#pod BatchChunker will still use the initial C, and it will need at least one
#pod chunk processed, before it makes adjustments.  If the starting chunk size is grossly
#pod inaccurate to the workload, you could end up with several chunks in the beginning causing
#pod long-lasting locks before the runtime targeting reduces them down to a reasonable size.
#pod
#pod (Chunk size reductions are prioritized before increases, so it should re-size as soon as
#pod it finds the problem.  But, one bad chunk could be all it takes to cause an outage.)
#pod
#pod Default is 5 seconds.  Set this to zero to turn off runtime targeting.  (This was
#pod previously defaulted to off prior to v0.92, and set to 15 in v0.92.)
#pod
#pod =cut

has target_time => (
    is       => 'ro',
    isa      => PositiveOrZeroNum,
    required => 0,
    default  => 5,
);

#pod =head3 sleep
#pod
#pod The number of seconds to sleep after each chunk.  It uses L's version, so
#pod fractional numbers are allowed.
#pod
#pod Default is 0.5 seconds, which is fine for most operations.  You can likely get away with
#pod zero for smaller operations, but test it out first.  If processing is going to take up a
#pod lot of disk I/O, you may want to consider a higher setting.  If the database server
#pod spends too much time on processing, the replicas may have a hard time keeping up with
#pod standard load.
#pod
#pod This will increase the overall processing time of the loop, so try to find a balance
#pod between the two.
#pod
#pod (This was previously defaulted to 0 seconds, prior to v1.0.0.)
#pod
#pod =cut

has 'sleep' => (
    is       => 'ro',
    isa      => PositiveOrZeroNum,
    required => 0,
    default  => 0.5,
);

#pod =head3 max_runtime
#pod
#pod The number of seconds that the entire process is allowed to run.  If you have a
#pod long-running I operation that you don't want to run for days, you can set
#pod this attribute, execute the operation, and run it again at a later date.  The L
#pod will be set to the last-used ID, so the operation can be continued with another
#pod L call.  Or you can use this to figure out if it finished or not.
#pod
#pod Turned off by default.  If you use this, you should add in extra multipler operations to
#pod separate out the time math, like C<6 * 60 * 60> for 6 hours.
#pod
#pod =cut

has max_runtime => (
    is       => 'ro',
    isa      => PositiveOrZeroNum,
    required => 0,
    default  => 0,
);

#pod =head3 process_past_max
#pod
#pod Boolean that controls whether to check past the L during the loop.  If the loop
#pod hits the end point, it will run another maximum ID check in the DB, and adjust C
#pod accordingly.  If it somehow cannot run a DB check (no L or L available,
#pod for example), the last chunk will just be one at the end of C<< max_id + chunk_size >>.
#pod
#pod This is useful if the entire table is expected to be processed, and you don't want to
#pod miss any new rows that come up between L and the end of the loop.
#pod
#pod Turned off by default.
#pod
#pod =cut

has process_past_max => (
    is       => 'ro',
    isa      => Bool,
    required => 0,
    default  => 0,
);

#pod =head3 single_rows
#pod
#pod Boolean that controls whether single rows are passed to the L or the chunk's
#pod ResultSets/statement handle is passed.
#pod
#pod Since running single-row operations in a DB is painfully slow (compared to bulk
#pod operations), this also controls whether the entire set of coderefs are encapsulated into
#pod a DB transaction.  Transactionalizing the entire chunk brings the speed, and atomicity,
#pod back to what a bulk operation would be.  (Bulk operations are still faster, but you can't
#pod do anything you want in a single DML statement.)
#pod
#pod Used only by L and L.
#pod
#pod =cut

has single_rows => (
    is       => 'ro',
    isa      => Bool,
    required => 0,
    default  => 0,
);

#pod =head3 min_chunk_percent
#pod
#pod The minimum row count, as a percentage of L.  This value is actually
#pod expressed in decimal form, i.e.: between 0 and 1.
#pod
#pod This value will be used to determine when to process, skip, or expand a block, based on
#pod a count query.  The default is C<0.5> or 50%, which means that it will try to expand the
#pod block to a larger size if the row count is less than 50% of the chunk size.  Zero-sized
#pod blocks will be skipped entirely.
#pod
#pod This "chunk resizing" is useful for large regions of the table that have been deleted, or
#pod when the incrementing ID has large gaps in it for other reasons.  Wasting time on
#pod numerical gaps that span millions can slow down the processing considerably, especially
#pod if L is enabled.
#pod
#pod If this needs to be disabled, set this to 0.  The maximum chunk percentage does not have
#pod a setting and is hard-coded at C<< 100% + min_chunk_percent >>.
#pod
#pod If DBIC processing isn't used, L is also required to enable chunk resizing.
#pod
#pod =cut

has min_chunk_percent => (
    is       => 'ro',
    isa      => Type::Utils::declare(
        name    => 'PositiveZeroToOneNum',
        parent  => NumRange->parameterize(0, 1),
        message => sub { 'Must be a number between 0 and 1' },
    ),
    required => 0,
    default  => 0.5,
);

#pod =head3 min_id
#pod
#pod =head3 max_id
#pod
#pod Used by L to figure out the main start and end points.  Calculated by
#pod L.
#pod
#pod Manually setting this is not recommended, as each database is different and the
#pod information may have changed between the DB change development and deployment.  Instead,
#pod use L to fill in these values right before running the loop.
#pod
#pod When the operation is finished, C will be set to the last processed ID, just in
#pod case it was stopped early and needs to be restarted (eg: L is set).
#pod Alternately, you can run L again to confirm from the database.
#pod
#pod =cut

has min_id => (
    is       => 'rw',
    isa      => UnsignedInt,
);

has max_id => (
    is       => 'rw',
    isa      => UnsignedInt,
);

# Big number handling
has _use_bignums => (
    is       => 'rw',
    isa      => Bool,
    default  => 0,
    trigger  => \&_upgrade_attrs_to_bigint,
);

my @BIGNUM_BC_ATTRS = (qw< chunk_size min_id max_id >);
my @BIGNUM_LS_ATTRS = (qw< start end prev_end multiplier_range multiplier_step chunk_size chunk_count >);

sub _check_bignums {
    my ($self) = shift;
    return 1 if $self->_use_bignums;  # already checked these

    # Auto-set _use_bignums if we detect that we need it
    my $set_bignums = 0;

    # If other values are passed, check those, too
    foreach my $val (@_) {
        next unless defined $val;
        $set_bignums = 1 if blessed $val || !PerlSafeInt->check($val);
    }

    # Check BatchChunker attributes
    foreach my $attr (@BIGNUM_BC_ATTRS) {
        my $val = $self->$attr();
        next unless defined $val;
        $set_bignums = 1 if blessed $val || !PerlSafeInt->check($val);
    }

    # Check LoopState attributes
    if (my $ls = $self->loop_state) {
        foreach my $attr (@BIGNUM_LS_ATTRS) {
            my $val = $ls->$attr();
            next unless defined $val;
            $set_bignums = 1 if blessed $val || !PerlSafeInt->check($val);
        }
    }

    $self->_use_bignums(1) if $set_bignums;
    return $set_bignums;
}

sub _upgrade_attrs_to_bigint {
    my ($self, $is_on) = @_;
    return unless $is_on;

    # Fix BatchChunker attributes
    foreach my $attr (@BIGNUM_BC_ATTRS) {
        my $val = $self->$attr();
        next unless defined $val;  # nothing to upgrade
        next if blessed $val;      # already upgraded
        $self->$attr( Math::BigInt->new($val) );
    }

    # Fix LoopState attributes
    my $ls = $self->loop_state;
    return unless $ls;
    foreach my $attr (@BIGNUM_LS_ATTRS) {
        my $val = $ls->$attr();
        next unless defined $val;  # nothing to upgrade
        next if blessed $val;      # already upgraded
        $ls->$attr( Math::BigInt->new($val) );
    }
}

#pod =head3 loop_state
#pod
#pod A L object designed to hold variables during the
#pod processing loop.  The object will be cleared out after use.  Most of the complexity is
#pod needed for chunk resizing.
#pod
#pod =cut

has loop_state => (
    is       => 'rw',
    isa      => InstanceOf['DBIx::BatchChunker::LoopState'],
    required => 0,
    init_arg => undef,
    clearer  => 'clear_loop_state',
);

# Backwards-compatibility
*_loop_state = \&loop_state;

#pod =for Pod::Coverage BUILDARGS BUILD
#pod
#pod =cut

around BUILDARGS => sub {
    my $next  = shift;
    my $class = shift;

    my %args = @_ == 1 ? %{ $_[0] } : @_;

    # debug -> verbose
    $args{verbose} //= delete $args{debug} if exists $args{debug};

    # Auto-building of rsc and id_name can be a weird dependency dance, so it's better to
    # handle it here.
    my ($rsc, $rs, $id_name) = @args{qw< rsc rs id_name >};
    if    (defined $rsc && !$id_name) {
        $args{id_name} = $rsc->{_as};
    }
    elsif (!defined $rsc && $id_name && defined $rs) {
        $args{rsc}     = $rs->get_column( $args{id_name} );
    }
    elsif (!defined $rsc && !$id_name && defined $rs) {
        $args{id_name} = ($rs->result_source->primary_columns)[0];
        $args{rsc}     = $rs->get_column( $args{id_name} );
    }
    $rsc = $args{rsc};

    # Default count_rs is just $rs
    $args{count_rs} //= $rs if defined $rs;

    # Auto-add dbic_storage, if available
    if (!defined $args{dbic_storage} && (defined $rs || defined $rsc)) {
        $args{dbic_storage} = defined $rs ? $rs->result_source->storage : $rsc->_resultset->result_source->storage;
    }

    # Find something to use as a dbi_connector, if it doesn't already exist
    my @old_attrs = qw< sth min_sth max_sth count_sth >;
    my @new_attrs = map { my $k = $_; $k =~ s/sth$/stmt/; $k } @old_attrs;
    my $example_key = first { $args{$_} } @old_attrs;
    if ($example_key && !defined $args{dbi_connector}) {
        warn join "\n",
            'The sth/*_sth options are now considered legacy usage in DBIx::BatchChunker.  Because there is no',
            'way to re-acquire the password, any attempt to reconnect will fail.  Please use dbi_connector and',
            'stmt/*_stmt instead for reconnection support.',
            ''
        ;

        # NOTE: There was a way to monkey-patch _connect to use $dbh->clone, but I've considered it
        # too intrusive of a solution to use.  Better to demand that the user switch to the new
        # attributes, but have something that still works in most cases.

        # Attempt to build some sort of Connector object
        require DBIx::Connector::Retry;
        my $dbh = $args{$example_key}->{Database};

        my $conn = DBIx::Connector::Retry->new(
            connect_info => [
                join(':', 'dbi', $dbh->{Driver}{Name}, $dbh->{Name}),
                $dbh->{Username},
                '',  # XXX: Can't acquire the password
                # Sane %attr defaults on the off-chance that it actually re-connects
                { AutoCommit => 1, RaiseError => 1 },
            ],

            # Do not disconnect on DESTROY.  The $dbh might still be used post-run.
            disconnect_on_destroy => 0,
        );

        # Pretend $conn->_connect was called and store our pre-existing $dbh
        $conn->{_pid} = $$;
        $conn->{_tid} = threads->tid if $INC{'threads.pm'};
        $conn->{_dbh} = $dbh;
        $conn->driver;

        $args{dbi_connector} = $conn;
    }

    # Handle legacy options for sth/*_sth
    foreach my $old_attr (grep { $args{$_} } @old_attrs) {
        my $new_attr = $old_attr;
        $new_attr =~ s/sth$/stmt/;

        my $sth = delete $args{$old_attr};
        $args{$new_attr} ||= [ $sth->{Statement} ];
    }

    # Now check to make sure dbi_connector is available for DBI processing
    die 'DBI processing requires a dbi_connector or dbic_storage attribute!' if (
        !(defined $args{dbi_connector} || defined $args{dbic_storage}) &&
        (defined first { $args{$_} } @new_attrs)
    );

    # Other sanity checks
    die 'Range calculations require one of these attr sets: rsc, rs, or dbi_connector|dbic_storage + min_stmt + max_stmt' unless (
        defined $args{rsc} ||
        (defined $args{min_stmt} && defined $args{max_stmt}) ||
        (!defined $args{dbi_connector} && !defined $args{dbic_storage} && defined $args{coderef})  # DIY mode is exempt
    );

    die 'Block execution requires one of these attr sets: dbi_connector|dbic_storage + stmt, rs + coderef, or coderef' unless (
        $args{stmt} ||
        (defined $args{rs} && $args{coderef}) ||
        $args{coderef}
    );

    if (exists $args{target_time} && $args{target_time} == 0 && !$args{chunk_size}) {
        warn join "\n",
            'Dynamic chunk resizing is turned off and the chunk_size is still set to its default of 1.',
            'This is probably not desirable, and you should find an appropriate static chunk size for',
            'your workload.',
            ''
        ;
    }

    $class->$next( %args );
};

sub BUILD {
    my $self = shift;
    # Make sure id_name gets fixed at the right time
    $self->_fix_id_name( $self->id_name );
    $self->_check_bignums;
}

#pod =head1 CONSTRUCTORS
#pod
#pod See L for information on what can be passed into these constructors.
#pod
#pod =head2 new
#pod
#pod     my $batch_chunker = DBIx::BatchChunker->new(...);
#pod
#pod A standard object constructor. If you use this constructor, you will need to
#pod manually call L and L to execute the DB changes.
#pod
#pod =head2 construct_and_execute
#pod
#pod     my $batch_chunker = DBIx::BatchChunker->construct_and_execute(...);
#pod
#pod Constructs a DBIx::BatchChunker object and automatically calls
#pod L and L on it. Anything passed to this method will be passed
#pod through to the constructor.
#pod
#pod Returns the constructed object, post-execution.  This is typically only useful if you want
#pod to inspect the attributes after the process has finished.  Otherwise, it's safe to just
#pod ignore the return and throw away the object immediately.
#pod
#pod =cut

sub construct_and_execute {
    my $class     = shift;
    my $db_change = $class->new(@_);

    $db_change->calculate_ranges;
    $db_change->execute;

    return $db_change;
}

#pod =head1 METHODS
#pod
#pod =head2 calculate_ranges
#pod
#pod     my $batch_chunker = DBIx::BatchChunker->new(
#pod         rsc      => $account_rsc,    # a ResultSetColumn
#pod         ### OR ###
#pod         rs       => $account_rs,     # a ResultSet
#pod         id_name  => 'account_id',    # can be looked up if not provided
#pod         ### OR ###
#pod         dbi_connector => $conn,      # DBIx::Connector::Retry object
#pod         min_stmt      => $min_stmt,  # a SQL statement or DBI $sth args
#pod         max_stmt      => $max_stmt,  # ditto
#pod
#pod         ### Optional but recommended ###
#pod         id_name      => 'account_id',  # will also be added into the progress bar title
#pod         chunk_size   => 20_000,        # default is 1000
#pod
#pod         ### Optional ###
#pod         progress_bar => $progress,     # defaults to a 2-count 'Calculating ranges' bar
#pod
#pod         # ...other attributes for execute...
#pod     );
#pod
#pod     my $has_data_to_process = $batch_chunker->calculate_ranges;
#pod
#pod Given a L, L, or L statement
#pod argument set, this method calculates the min/max IDs of those objects.  It fills in the
#pod L and L attributes, based on the ID data, and then returns 1.
#pod
#pod If either of the min/max statements don't return any ID data, this method will return 0.
#pod
#pod =cut

sub calculate_ranges {
    my $self = shift;

    my $column_name = $self->id_name || '';
    $column_name =~ s/^\w+\.//;

    my $progress = $self->progress_bar || Term::ProgressBar->new({
        name   => 'Calculating ranges'.($column_name ? " for $column_name" : ''),
        count  => 2,
        ETA    => 'linear',
        silent => !(-t *STDERR && -t *STDIN),  # STDERR is what {fh} is set to use
    });

    # Actually run the statements
    my ($min_id, $max_id);
    if ($self->rsc) {
        $self->_dbic_block_runner( run => sub {
            # In case the sub is retried
            $progress->update(0);

            $min_id = $self->rsc->min;
            $progress->update(1);

            $max_id = $self->rsc->max;
            $progress->update(2);
        });
    }
    elsif ($self->dbic_storage) {
        $self->_dbic_block_runner( run => sub {
            my $dbh = $self->dbic_storage->dbh;

            # In case the sub is retried
            $progress->update(0);

            ($min_id) = $dbh->selectrow_array(@{ $self->min_stmt });
            $progress->update(1);

            ($max_id) = $dbh->selectrow_array(@{ $self->max_stmt });
            $progress->update(2);
        });
    }
    else {
        $self->dbi_connector->run(sub {
            my $dbh = $_;

            # In case the sub is retried
            $progress->update(0);

            ($min_id) = $dbh->selectrow_array(@{ $self->min_stmt });
            $progress->update(1);

            ($max_id) = $dbh->selectrow_array(@{ $self->max_stmt });
            $progress->update(2);
        });
    }

    # Set the ranges and return
    return 0 unless defined $min_id && defined $max_id;

    # This would be the primary spot where we notice we need to upgrade, so check the values before
    # we attempt to mangle them.
    if ($self->_check_bignums($min_id, $max_id)) {
        $min_id = Math::BigFloat->new($min_id)->as_int;
        $max_id = Math::BigFloat->new($max_id)->as_int;
    }
    else {
        $min_id = int $min_id;
        $max_id = int $max_id;
    }

    $self->min_id($min_id);
    $self->max_id($max_id);

    return 1;
}

#pod =head2 execute
#pod
#pod     my $batch_chunker = DBIx::BatchChunker->new(
#pod         # ...other attributes for calculate_ranges...
#pod
#pod         dbi_connector => $conn,          # DBIx::Connector::Retry object
#pod         stmt          => $do_stmt,       # INSERT/UPDATE/DELETE $stmt with BETWEEN placeholders
#pod         ### OR ###
#pod         dbi_connector => $conn,          # DBIx::Connector::Retry object
#pod         stmt          => $select_stmt,   # SELECT $stmt with BETWEEN placeholders
#pod         count_stmt    => $count_stmt,    # SELECT COUNT $stmt to be used for min_chunk_percent; optional
#pod         coderef       => $coderef,       # called code that does the actual work
#pod         ### OR ###
#pod         rs      => $account_rs,          # base ResultSet, which gets filtered with -between later on
#pod         id_name => 'account_id',         # can be looked up if not provided
#pod         coderef => $coderef,             # called code that does the actual work
#pod         ### OR ###
#pod         coderef => $coderef,             # DIY database work; just pass the $start/$end IDs
#pod
#pod         ### Optional but recommended ###
#pod         sleep             => 0.25, # number of seconds to sleep each chunk; defaults to 0
#pod         process_past_max  => 1,    # use this if processing the whole table
#pod         single_rows       => 1,    # does $coderef get a single $row or the whole $chunk_rs / $stmt
#pod         min_chunk_percent => 0.25, # minimum row count of chunk size percentage; defaults to 0.5 (or 50%)
#pod         target_time       => 5,    # target runtime for dynamic chunk size scaling; default is 5 seconds
#pod         max_runtime       => 12 * 60 * 60, # stop processing after 12 hours
#pod
#pod         progress_name => 'Updating Accounts',  # easier than creating your own progress_bar
#pod
#pod         ### Optional ###
#pod         progress_bar     => $progress,  # defaults to "Processing $source_name" bar
#pod         verbose          => 1,          # displays timing stats on each chunk
#pod     );
#pod
#pod     $batch_chunker->execute if $batch_chunker->calculate_ranges;
#pod
#pod Applies the configured DB changes in chunks.  Runs through the loop, processing a
#pod statement handle, ResultSet, and/or coderef as it goes.  Each loop iteration processes a
#pod chunk of work, determined by L.
#pod
#pod The L method should be run first to fill in L and L.
#pod If either of these are missing, the function will assume L couldn't
#pod find them and warn about it.
#pod
#pod More details can be found in the L and L sections.
#pod
#pod =cut

sub execute {
    my $self = shift;
    $self->_check_bignums;

    my $count;
    if (defined $self->min_id && defined $self->max_id) {
        $count = $self->max_id - $self->min_id + 1;
    }

    # Fire up the progress bar
    my $progress = $self->progress_bar || Term::ProgressBar->new({
        name   => $self->progress_name,
        count  => $count || 1,
        ETA    => 'linear',
        silent => !(-t *STDERR && -t *STDIN),  # STDERR is what {fh} is set to use
    });

    unless ($count) {
        $progress->message('No chunks; nothing to process...');
        return;
    }

    if ($self->verbose) {
        $progress->message(
            sprintf "(%s total chunks; %s total rows)",
                map { $self->cldr->decimal_formatter->format($_) } ( ceil($count / $self->chunk_size), $count)
        );
    }

    # Loop state setup
    $self->clear_loop_state;
    my $ls = $self->loop_state( DBIx::BatchChunker::LoopState->new({
        batch_chunker => $self,
        progress_bar  => $progress,
    }) );

    # Da loop
    while (
        !defined $ls->prev_end ||      # first chunk
        defined  $ls->start    ||      # still in the middle of chunk resizing
        $ls->prev_end < $self->max_id  # processed chunk, but still more to go
    ) {
        $ls->multiplier_range($ls->multiplier_range + $ls->multiplier_step);

        # this could be already set, if chunk resizing
        $ls->start(
            defined $ls->prev_end ? $ls->prev_end + 1 :
            defined $ls->min_id   ? $ls->min_id       :
            1
        ) unless defined $ls->start;

        $ls->end(
            min(
                $ls->start + ceil($ls->multiplier_range * $ls->chunk_size) - 1, # ceil, because multiplier_* could be fractional
                $self->max_id,  # ensure we never exceed max_id
            )
        );
        $ls->chunk_count(undef);

        # Early loop exit because of maximum run time
        if ($self->max_runtime && time - $ls->total_timer > $self->max_runtime) {
            $progress->message('Ran past the maximum run time');
            last;
        }

        next unless $self->_process_past_max_checker;

        # The actual DB processing
        next unless $self->_process_block;

        # Record the time quickly
        $ls->prev_runtime(time - $ls->chunk_timer);

        # Give the DB a little bit of breathing room
        sleep $self->sleep if $self->sleep;

        $self->_print_chunk_status('processed');
        $self->_increment_progress;
        $self->_runtime_checker;

        # End-of-loop activities (skipped by early next)
        $ls->_reset_chunk_state;
    }

    # Re-set min_id and clear the loop state
    $self->min_id( $ls->prev_end ) if defined $ls->prev_end;
    $self->clear_loop_state;

    # Keep the finished time from the progress bar, in case there are other loops or output
    unless ($progress->silent) {
        $progress->update( $progress->target );
        print "\n";
    }
}

#pod =head1 PRIVATE METHODS
#pod
#pod =head2 _process_block
#pod
#pod Runs the DB work and passes it to the coderef.  Its return value determines whether the
#pod block should be processed or not.
#pod
#pod =cut

sub _process_block {
    my ($self) = @_;

    my $ls       = $self->loop_state;
    my $conn     = $self->dbi_connector;
    my $coderef  = $self->coderef;
    my $rs       = $self->rs;
    my $count_rs = $self->count_rs // $rs;

    # Figure out if the row count is worth the work
    my $chunk_rs;
    my $count_stmt = $self->count_stmt;
    my $chunk_count;
    if ($count_stmt && defined $self->dbic_storage) {
        $self->_dbic_block_runner( run => sub {
            $chunk_count = $self->dbic_storage->dbh->selectrow_array(
                @$count_stmt,
                (@$count_stmt == 1 ? undef : ()),
                $ls->start, $ls->end,
            );
        });
    }
    elsif ($count_stmt) {
        $chunk_count = $conn->run(sub {
            $_->selectrow_array(
                @$count_stmt,
                (@$count_stmt == 1 ? undef : ()),
                $ls->start, $ls->end,
            );
        });
    }
    elsif (defined $count_rs) {
        $chunk_rs = $count_rs->search({
            $self->id_name => { -between => [$ls->start, $ls->end] },
        });

        $self->_dbic_block_runner( run => sub {
            $chunk_count = $chunk_rs->count;
        });
    }

    $chunk_count = Math::BigInt->new($chunk_count) if $self->_check_bignums($chunk_count);
    $ls->chunk_count($chunk_count);

    return unless $self->_chunk_count_checker;

    # NOTE: Try to minimize the amount of closures by using $self as much as possible
    # inside coderefs.

    # Do the work
    if (my $stmt = $self->stmt) {
        ### Statement handle
        my @prepare_args = @$stmt > 2 ? @$stmt[0..1] : @$stmt;
        my @execute_args = (
            (@$stmt > 2 ? @$stmt[2..$#$stmt] : ()),
            $ls->start, $ls->end,
        );

        if ($self->single_rows && $coderef) {
            # Transactional work
            if ($self->dbic_storage) {
                $self->_dbic_block_runner( txn => sub {
                    $self->loop_state->_mark_chunk_timer;  # reset timer on retries

                    my $sth = $self->dbic_storage->dbh->prepare(@prepare_args);
                    $sth->execute(@execute_args);

                    while (my $row = $sth->fetchrow_hashref('NAME_lc')) { $self->coderef->($self, $row) }
                });
            }
            else {
                $conn->txn(sub {
                    $self->loop_state->_mark_chunk_timer;  # reset timer on retries

                    my $sth = $_->prepare(@prepare_args);
                    $sth->execute(@execute_args);

                    while (my $row = $sth->fetchrow_hashref('NAME_lc')) { $self->coderef->($self, $row) }
                });
            }
        }
        else {
            # Bulk work (or DML)
            if ($self->dbic_storage) {
                $self->_dbic_block_runner( run => sub {
                    $self->loop_state->_mark_chunk_timer;  # reset timer on retries

                    my $sth = $self->dbic_storage->dbh->prepare(@prepare_args);
                    $sth->execute(@execute_args);

                    $self->coderef->($self, $sth) if $self->coderef;
                });
            }
            else {
                $conn->run(sub {
                    $self->loop_state->_mark_chunk_timer;  # reset timer on retries

                    my $sth = $_->prepare(@prepare_args);
                    $sth->execute(@execute_args);

                    $self->coderef->($self, $sth) if $self->coderef;
                });
            }
        }
    }
    elsif (defined $rs && $coderef) {
        ### ResultSet with coderef

        if ($self->single_rows) {
            # Transactional work
            $self->_dbic_block_runner( txn => sub {
                # reset timer/$rs on retries
                $self->loop_state->_mark_chunk_timer;
                $chunk_rs->reset;

                while (my $row = $chunk_rs->next) { $self->coderef->($self, $row) }
            });
        }
        else {
            # Bulk work
            $self->_dbic_block_runner( run => sub {
                # reset timer/$rs on retries
                $self->loop_state->_mark_chunk_timer;
                $chunk_rs->reset;

                $self->coderef->($self, $chunk_rs);
            });
        }
    }
    else {
        ### Something a bit more free-form

        $self->$coderef($ls->start, $ls->end);
    }

    return 1;
}

#pod =head2 _process_past_max_checker
#pod
#pod Checks to make sure the current endpoint is actually the end, by checking the database.
#pod Its return value determines whether the block should be processed or not.
#pod
#pod See L.
#pod
#pod =cut

sub _process_past_max_checker {
    my ($self) = @_;
    my $ls = $self->loop_state;
    my $progress = $ls->progress_bar;

    return 1 unless $self->process_past_max;
    return 1 unless $ls->end >= $self->max_id;

    # No checks for DIY, if they didn't include a max_stmt
    unless (defined $self->rsc || $self->max_stmt) {
        # There's no way to size this, so add one more chunk
        $ls->end($self->max_id + $ls->chunk_size);
        return 1;
    }

    # Run another MAX check
    $progress->message('Reached end; re-checking max ID') if $self->verbose;
    my $new_max_id;
    if (defined( my $rsc = $self->rsc )) {
        $self->_dbic_block_runner( run => sub {
            $new_max_id = $rsc->max;
        });
    }
    elsif ($self->dbic_storage) {
        $self->_dbic_block_runner( run => sub {
            ($new_max_id) = $self->dbic_storage->dbh->selectrow_array(@{ $self->max_stmt });
        });
    }
    else {
        ($new_max_id) = $self->dbi_connector->run(sub {
            $_->selectrow_array(@{ $self->max_stmt });
        });
    }
    $ls->_mark_chunk_timer;  # the above query shouldn't impact runtimes

    # Convert $new_max_id if necessary
    $new_max_id = Math::BigInt->new($new_max_id) if $self->_check_bignums($new_max_id);

    if (!$new_max_id || $new_max_id eq '0E0') {
        # No max: No affected rows to change
        $progress->message('No max ID found; nothing left to process...') if $self->verbose;
        $ls->end($self->max_id);

        $ls->prev_check('no max');
        return 0;
    }
    elsif ($new_max_id > $self->max_id) {
        # New max ID
        $progress->message( sprintf 'New max ID set from %s to %s', $self->max_id, $new_max_id ) if $self->verbose;
        $self->max_id($new_max_id);
        $progress->target( $new_max_id - $self->min_id + 1 );
        $progress->update( $progress->last_update );
    }
    elsif ($new_max_id == $self->max_id) {
        # Same max ID
        $progress->message( sprintf 'Found max ID %s; same as end', $new_max_id ) if $self->verbose;
    }
    else {
        # Max too low
        $progress->message( sprintf 'Found max ID %s; ignoring...', $new_max_id ) if $self->verbose;
    }

    # Run another boundary check with the new max_id value
    $ls->end( min($ls->end, $self->max_id) );

    return 1;
}

#pod =head2 _chunk_count_checker
#pod
#pod Checks the chunk count to make sure it's properly sized.  If not, it will try to shrink
#pod or expand the current chunk (in C increments) as necessary.  Its return value
#pod determines whether the block should be processed or not.
#pod
#pod See L.
#pod
#pod This is not to be confused with the L, which adjusts C
#pod after processing, based on previous run times.
#pod
#pod =cut

sub _chunk_count_checker {
    my ($self) = @_;
    my $ls = $self->loop_state;
    my $progress = $ls->progress_bar;

    # Chunk sizing is essentially disabled, so bounce out of here
    if ($self->min_chunk_percent <= 0 || !defined $ls->chunk_count) {
        $ls->prev_check('disabled');
        return 1;
    }

    my $chunk_percent    = $ls->chunk_count / $ls->chunk_size;
    my $count_check_time = time - $ls->chunk_timer;  # should only include the COUNT time at this point
    $ls->checked_count( $ls->checked_count + 1 );

    if    ($ls->chunk_count == 0 && $self->min_chunk_percent > 0) {
        # No rows: Skip the block entirely, and accelerate the stepping
        $self->_print_chunk_status('skipped');

        $self->_increment_progress;

        my $step = $ls->multiplier_step;
        $ls->_reset_chunk_state;
        $ls->multiplier_step( $step * 2 );

        $ls->prev_check('skipped rows');
        return 0;
    }
    elsif ($ls->end - $ls->start <= 0) {
        # Down to a single ID: We _have_ to process it
        $ls->prev_check('at a single ID');

        # Complain, because this can be dangerous with a wild enough Row:ID ratio
        if ($ls->chunk_count > 1) {
            $progress->message('WARNING: Processing a single ID with many rows attached because resizing cannot proceed any further.');
            $progress->message('Consider flipping the relationship so that IDs and row counts are 1:1.');
        }

        return 1;
    }
    elsif ($chunk_percent > 1 + $self->min_chunk_percent) {
        # Too many rows: Backtrack to the previous range and try to bisect
        $self->_print_chunk_status('shrunk');
        $ls->_mark_chunk_timer;
        $ls->_decrease_multiplier;
        $ls->prev_check('too many rows');
        return 0;
    }
    elsif ($self->target_time && $count_check_time > $self->target_time * 1.05) {
        # COUNT statement too slow: Backtrack to the previous range and try to bisect

        # This is a rare failure, so print a warning
        my $integer = $self->cldr->decimal_formatter;
        my $decimal = $self->cldr->decimal_formatter(
            minimum_fraction_digits => 2,
            maximum_fraction_digits => 2,
        );
        $progress->message( sprintf(
            'WARNING: COUNT statement was too slow; took %5s sec to return %s rows.',
            $decimal->format($count_check_time),
            $integer->format( $ls->chunk_count )
        ) );

        $self->_print_chunk_status('shrunk');
        $ls->_mark_chunk_timer;
        $ls->_decrease_multiplier;
        $ls->prev_check('COUNT too slow');
        return 0;
    }

    # The above four are more important than skipping the count checks.  Better to
    # have too few rows than too many.  The single ID check prevents infinite loops
    # from bisecting, though.

    elsif ($ls->checked_count > 10) {
        # Checked too many times: Just process it
        $ls->prev_check('too many checks');
        return 1;
    }
    elsif ($ls->end >= $self->max_id) {
        # At the end: Just process it
        $ls->prev_check('at max_id');
        return 1;
    }
    elsif ($chunk_percent < $self->min_chunk_percent) {
        # Too few rows: Keep the start ID and accelerate towards a better endpoint
        $self->_print_chunk_status('expanded');
        $ls->_mark_chunk_timer;
        $ls->_increase_multiplier;
        $ls->prev_check('too few rows');
        return 0;
    }

    $ls->prev_check('nothing wrong');
    return 1;
}

#pod =head2 _runtime_checker
#pod
#pod Stores the previously processed chunk's runtime, and then adjusts C as
#pod necessary.
#pod
#pod See L.
#pod
#pod =cut

sub _runtime_checker {
    my ($self) = @_;
    my $ls = $self->loop_state;
    return unless $self->target_time;
    return unless $ls->chunk_size && $ls->prev_runtime;  # prevent DIV/0

    my $timings = $ls->last_timings;

    my $new_timing = {
        runtime     => $ls->prev_runtime,
        chunk_count => $ls->chunk_count || $ls->chunk_size,
    };
    $new_timing->{chunk_per} = $new_timing->{chunk_count} / $ls->chunk_size;

    # Rowtime: a measure of how much of the chunk_size actually impacted the runtime
    $new_timing->{rowtime} = $new_timing->{runtime} / $new_timing->{chunk_per};

    # Store the last five processing times
    push @$timings, $new_timing;
    shift @$timings if @$timings > 5;

    # Figure out the averages and adjustment factor
    my $ttl = scalar @$timings;
    my $avg_rowtime   = sum(map { $_->{rowtime} } @$timings) / $ttl;
    my $adjust_factor = $self->target_time / $avg_rowtime;

    my $new_target_chunk_size = $ls->chunk_size;
    my $adjective;
    if    ($adjust_factor > 1.05) {
        # Too fast: Raise the chunk size

        return unless $ttl >= 5;                                          # must have a full set of timings
        return if any { $_->{runtime} >= $self->target_time } @$timings;  # must ALL have low runtimes

        $new_target_chunk_size *= min(2, $adjust_factor);  # never more than double
        $adjective = 'fast';
    }
    elsif ($adjust_factor < 0.95) {
        # Too slow: Lower the chunk size

        return unless $ls->prev_runtime > $self->target_time;  # last runtime must actually be too high

        $new_target_chunk_size *=
            ($ls->prev_runtime < $self->target_time * 3) ?
            max(0.5, $adjust_factor) :  # never less than half...
            $adjust_factor              # ...unless the last runtime was waaaay off
        ;
        $new_target_chunk_size = 1 if $new_target_chunk_size < 1;
        $adjective = 'slow';
    }

    $new_target_chunk_size = int $new_target_chunk_size;
    return if $new_target_chunk_size == $ls->chunk_size;  # either nothing changed or it's too miniscule
    return if $new_target_chunk_size < 1;

    # Print out a processing line, if enabled
    if ($self->verbose) {
        # CLDR number formatters
        my $integer = $self->cldr->decimal_formatter;
        my $percent = $self->cldr->percent_formatter;

        $ls->{progress_bar}->message( sprintf(
            "Processing too %s, avg %4s of target time, adjusting chunk size from %s to %s",
            $adjective,
            $percent->format( 1 / $adjust_factor ),
            $integer->format( $ls->chunk_size ),
            $integer->format( $new_target_chunk_size ),
        ) );
    }

    # Change it!
    $ls->chunk_size($new_target_chunk_size);
    $ls->_reset_last_timings if $adjective eq 'fast';  # never snowball too quickly
    return 1;
}

#pod =head2 _increment_progress
#pod
#pod Increments the progress bar.
#pod
#pod =cut

sub _increment_progress {
    my ($self) = @_;
    my $ls = $self->loop_state;
    my $progress = $ls->progress_bar;

    my $so_far = $ls->end - $self->min_id + 1;
    $progress->target($so_far+1) if $ls->end > $self->max_id;
    $progress->update($so_far);
}

#pod =head2 _print_chunk_status
#pod
#pod Prints out a standard chunk status line, if L is enabled.  What it prints is
#pod generally uniform, but it depends on the processing action.  Most of the data is
#pod pulled from L.
#pod
#pod =cut

sub _print_chunk_status {
    my ($self, $action) = @_;
    return unless $self->verbose;

    my $ls    = $self->loop_state;
    my $sleep = $self->sleep || 0;

    # CLDR number formatters
    my $integer = $self->cldr->decimal_formatter;
    my $percent = $self->cldr->percent_formatter;
    my $decimal = $self->cldr->decimal_formatter(
        minimum_fraction_digits => 2,
        maximum_fraction_digits => 2,
    );

    my $message;
    if ($ls->start < 1_000_000_000 && $ls->end < 1_000_000_000) {
        $message = sprintf(
            'IDs %6u to %6u %9s, %9s rows found',
            $ls->start, $ls->end, $action,
            $integer->format( $ls->chunk_count ),
        );
    }
    else {
        $message = sprintf(
            'IDs %s to %s %s, %s rows found',
            $ls->start, $ls->end, $action,
            $ls->chunk_count,
        );
    }

    $message .= sprintf(
        ' (%4s of chunk size)',
        $percent->format( $ls->chunk_count / $ls->chunk_size ),
    ) if $ls->chunk_count;

    if ($action eq 'processed') {
        $message .= $sleep ?
            sprintf(
                ', %5s+%s sec runtime+sleep',
                $decimal->format( $ls->prev_runtime ),
                $decimal->format( $sleep )
            ) :
            sprintf(
                ', %5s sec runtime',
                $decimal->format( $ls->prev_runtime ),
            )
        ;
    }

    # Reduce spacing if the numbers are too large
    if ($ls->start > 1_000_000_000 || $ls->end > 1_000_000_000) {
        $message =~ s/\s+/ /g;
        $message =~ s/\(\s+/\(/g;
    }

    return $ls->progress_bar->message($message);
}

#pod =head1 CAVEATS
#pod
#pod =head2 Big Number Support
#pod
#pod If the module detects that the ID numbers are no longer safe for standard Perl NV
#pod storage, it will automatically switch to using L and L for
#pod big number support.  If any blessed numbers are already being used to define the
#pod attributes, this will also switch on the support.
#pod
#pod =head2 String-based IDs
#pod
#pod If you're working with C types or other string-based IDs to represent integers,
#pod these may be subject to whatever string-based comparison rules your RDBMS uses when
#pod calculating with C/C or using C.  Row counting and chunk size scaling
#pod will try to compensate, but will be mixing string-based comparisons from the RDBMS and
#pod Perl-based integer math.
#pod
#pod Using the C function may help, but it may also cause critical indexes to be
#pod ignored, especially if the function is used on the left-hand side against the column.
#pod Strings with the exact same length may be safe from comparison weirdness, but YMMV.
#pod
#pod Non-integer inputs from ID columns, such as GUIDs or other alphanumeric strings, are not
#pod currently supported.  They would have to be converted to integers via SQL, and doing so
#pod may run into a similar risk of having your RDBMS ignore indexes.
#pod
#pod =head1 SEE ALSO
#pod
#pod L, L, L
#pod
#pod =cut

1;

__END__