File Coverage

blib/lib/BerkeleyDB/SecIndices/Accessor.pm

Criterion	Covered	Total	%
statement	13	15	86.6
branch			n/a
condition			n/a
subroutine	5	5	100.0
pod			n/a
total	18	20	90.0

package BerkeleyDB::SecIndices::Accessor;

use 5.005;
use strict;
use warnings;

require Exporter;

our @ISA = qw(Exporter);

our %EXPORT_TAGS = ( 'const' => [ qw(ELCK EPUT EDEL EUPD EGET EGTS
                                     EDUP EEPT ELOCK TRUE
) ] );

our @EXPORT_OK = ( @{ $EXPORT_TAGS{'const'} } );

our @EXPORT = qw(EGET EGTS EEPT EPUT ELCK EUPD);

our $VERSION = '0.06';

use Carp qw(croak);
use BerkeleyDB;
use File::Spec ();
use Storable qw(freeze thaw);
local $Storable::canonical = 1;

# global container to hold _ALL_ related stuff
my $STUBS = {};

# max retry times to obtain a cdb lock
our $LOCK_RETRY_MAX = 5;

# path to database configuration file
# of YAML format
our $CONFIG;
my $_config;

# configurable callbacks
# refer to BerkeleyDB on how to write
our $CB_EXTRACT_SECKEY;
our $CB_DUP;
our $CB_DUPSORT;

# debug flag
# no debug ouput by default
our $DEBUG = 0;

sub ELCK() { -2; }
sub ELOCK() { ELCK(); }
sub EPUT() { -1; }
sub EDEL() { -1; }
sub EUPD() { -1; }
sub EGET() { -1; }
sub EGTS() { -1; }
sub EDUP() { -1; }
sub EEPT() { -3; }
sub TRUE() { 1;  }

BEGIN {
    if ($BerkeleyDB::db_version < 3.3) {
        croak("BerkeleyDB ver 3.3.x required!");
    }
}

# This section inits _ALL_ databases
# _ALL_ db handlers are stored in $db_pool with specific keys
# the key is contructed according to keys %{$_config->{DATABASE}}

{
    use YAML ();
    ( $_config ) = YAML::LoadFile($CONFIG);
    # soft check
    croak("no database HOME found") unless
      exists $_config->{HOME} and -d $_config->{HOME};
    croak("wrong database declaration") unless
      ref $_config->{DATABASE} eq 'HASH';

    # Global Env for _ALL_ DBs
    # created in system shared memory by flag DB_SYSTEM_MEM
    # Shared memory key can also be specified in DB_CONFIG
    my $Env = BerkeleyDB::Env::->new(
        -Home         => $_config->{HOME},
        #-Cachesize    => 2000000,
        -ErrFile      => *STDERR,
        -ErrPrefix    => __PACKAGE__,
        -Flags        => DB_CREATE|DB_INIT_CDB|DB_INIT_MPOOL|DB_SYSTEM_MEM,
        -Verbose      => 1,
    );
    croak("cannot create dbenv") unless $Env and $Env->status() == 0;
    my $DB = $_config->{DATABASE};
    # container to hold  _ALL_ dbs
    my $db_pool = {};
    # container to check required index fields for primary database
    my $required_index = {};

    # DIRTY CODE START
    no strict 'refs';
    my $getenv = '___dbenv';

    *$getenv = sub() { $Env; };

    my $makekey = sub {
        my ( $primary, $index ) = @_;
        if ($index) {
            return '_'. lc($primary). '_'. lc($index);
        }
        else {
            return '_'. lc($primary);
        }
    };

    my $_db_property = DB_DUP;
    if (defined $CB_DUPSORT and ref $CB_DUPSORT eq 'CODE') {
        $_db_property |= DB_DUPSORT;
    }
    foreach my $i (keys %$DB) {
        if ($i =~ m/INDEX$/o) {
            # index db
            # open SUBS
            foreach my $j (0 .. $#{$DB->{$i}->{SUBS}}) {
                my $params = {
                    -Filename   => $DB->{$i}->{FILE},
                    -Subname    => $DB->{$i}->{SUBS}->[$j],
                    -Flags      => DB_CREATE|DB_DIRECT_DB,
                    -Property   => $_db_property,
                    -Mode       => 0644,
                    -Env        => $Env,
                };
                if (defined $CB_DUP and ref $CB_DUP eq 'CODE') {
                    $params->{-Compare} = $CB_DUP;
                }
                if (defined $CB_DUPSORT and
                      ref $CB_DUPSORT eq 'CODE') {
                    $params->{-DupCompare} = $CB_DUPSORT;
                }
                my $db = BerkeleyDB::Btree::->new(%$params);
                my $key = $makekey->($i, $DB->{$i}->{SUBS}->[$j]);
                #print 'key: ', $key, "\n";
                #print "CDS: ". $db->cds_enabled() ? 'true' : 'false', "\n";
                $db_pool->{$key} = $db;
                *$key = sub() { $db_pool->{$key}; };
            }
        }
        else {
            my $_type = 0; # Recno by default
            if (exists $DB->{$i}->{TYPE}) {
                # honor type
                #print STDERR $i, ": ", $DB->{$i}->{TYPE}, "\n";
                if ($DB->{$i}->{TYPE} eq 'Hash') {
                    $_type = 1;
                }
                elsif ($DB->{$i}->{TYPE} eq 'Btree') {
                    $_type = 2;
                }
                else {
                    $_type = 0;
                }
            }
            my $db;
            if ($_type == 0) {
                $db = BerkeleyDB::Recno::->new(
                    -Filename  => $DB->{$i}->{FILE},
                    -Flags     => DB_CREATE|DB_DIRECT_DB,
                    -Mode      => 0644,
                    -Env       => $Env,
                );
            }
            elsif ($_type == 1) {
                $db = BerkeleyDB::Hash::->new(
                    -Filename  => $DB->{$i}->{FILE},
                    -Flags     => DB_CREATE|DB_DIRECT_DB,
                    -Property  => DB_DUP,
                    -Mode      => 0644,
                    -Env       => $Env,
                );
            }
            else {
                $db = BerkeleyDB::Btree::->new(
                    -Filename  => $DB->{$i}->{FILE},
                    -Flags     => DB_CREATE|DB_DIRECT_DB,
                    -Property  => DB_DUP,
                    -Mode      => 0644,
                    -Env       => $Env,
                );
            }

            my $key = $makekey->($i);
            $db_pool->{$key} = $db;
            *$key = sub() { $db_pool->{$key}; };
        }
    }

    # associate each index db with primary db
    foreach my $i (keys %$DB) {
        if ($i =~ m/INDEX$/o) {
            # BLAH_INDEX associated with $db_pool->{_blah}
            foreach my $j (0 .. $#{$DB->{$i}->{SUBS}}) {
                my $secondary = $makekey->($i, $DB->{$i}->{SUBS}->[$j]);
                ( my $p = $i ) =~ s/_INDEX$//o;
                my $primary   = $makekey->($p);
                push @{$required_index->{$p}}, $DB->{$i}->{SUBS}->[$j];

                unless (exists $db_pool->{$primary}) {
                    croak("database $primary not found");
                }
                unless (exists $db_pool->{$secondary}) {
                    croak("database $secondary not found");
                }
                my $_extract_seckey = sub {
                    my $pkey  = shift;
                    # $pdata is freezed by Storable
                    my $pdata = shift;
                    my $hcontent = thaw($pdata);
                    my $k = $hcontent->{$DB->{$i}->{SUBS}->[$j]};
                    #print STDERR 'secondary key: '. $k, "\n";
                    if (ref $k eq 'ARRAY') {
                        # special index for array
                        # array items should be recno
                        # FOR scenarios and cases
                        my $skey = '';
                        foreach (@$k) {
                            vec($skey, $_, 1) = 1;
                        }
                        $_[0] = $skey;
                    }
                    else {
                        # normal SCALAR
                        $_[0] = $k;
                    }
                    return 0;
                };

                my $rc;
                if (defined $CB_EXTRACT_SECKEY and
                      ref $CB_EXTRACT_SECKEY eq 'CODE') {
                    $rc = $db_pool->{$primary}->associate(
                        $db_pool->{$secondary},
                        $CB_EXTRACT_SECKEY);
                }
                else {
                    $rc = $db_pool->{$primary}->associate(
                        $db_pool->{$secondary}, $_extract_seckey);
                }
                unless ($rc == 0) {
                    croak("cannot associate index $secondary ".
                            "with primary database $primary");
                }
            }
        }
    }

    # make accessor for each primary database
    foreach my $i (keys %$DB) {
        if ($i !~ m/INDEX$/o) {
            # primary database
            my $put   = 'put_'. lc($i);
            my $put2  = 'put2_'. lc($i);
            my $upd   = 'upd_'. lc($i);
            # TODO: update version 2
            my $upd2  = 'upd2_'. lc($i);
            my $get   = 'get_'. lc($i);
            my $gets  = 'get_'. lc($i). 's';
            my $count = '__'. lc($i). 's';
            my $dels  = 'del_'. lc($i). 's';
            $STUBS->{$i}->{PUT}   = sub { __PACKAGE__->$put(@_); };
            $STUBS->{$i}->{PUT2}  = sub { __PACKAGE__->$put2(@_); };
            $STUBS->{$i}->{UPD}   = sub { __PACKAGE__->$upd(@_); };
            #$STUBS->{$i}->{UPD2}  = sub { __PACKAGE__->$upd2(@_); };
            $STUBS->{$i}->{GET}   = sub { __PACKAGE__->$get(@_); };
            $STUBS->{$i}->{GETS}  = sub { __PACKAGE__->$gets(@_); };
            $STUBS->{$i}->{COUNT} = sub { __PACKAGE__->$count(@_); };
            $STUBS->{$i}->{DEL}   = sub { __PACKAGE__->$dels(@_); };

            if (exists $DB->{$i}->{TYPE} and
                  ($DB->{$i}->{TYPE} eq 'Hash' or
                     $DB->{$i}->{TYPE} eq 'Btree')) {
                *$put = sub {
                    my ( $self, $k, $hcontent ) = @_;
                    croak("ref $_[2] ne 'HASH'") unless
                      ref $hcontent eq 'HASH';
                    # check required index fields
                    foreach my $field (@{$required_index->{$i}}) {
                        if (not exists $hcontent->{$field}) {
                            croak("required index field $field ".
                                    "not found");
                        }
                    }
                    my $fcontent = freeze($hcontent);
                    my $key = $makekey->($i);
                    # validate $lock
                    my $lock;
                    my $retry = 0;
                    LOCK:
                    {
                        $lock = $db_pool->{$key}->cds_lock();
                        last LOCK if defined $lock;
                        sleep 3;
                        redo LOCK if ++$retry < $LOCK_RETRY_MAX;
                    }
                    unless (defined $lock) {
                        return ELCK;
                    }
                    my $rc = $db_pool->{$key}->db_put(
                        $k, $fcontent, DB_NOOVERWRITE);
                    # no cache in memo
                    $db_pool->{$key}->db_sync();
                    $lock->cds_unlock();
                    return $rc == 0 ? TRUE : EPUT;
                };

                *$put2 = sub {
                    my ( $self, $rpairs ) = @_;
                    return TRUE if keys %$rpairs == 0;

                    croak("ref $_[1] ne 'HASH'") unless
                      ref $rpairs eq 'HASH';
                    # check required index fields
                    foreach my $field (@{$required_index->{$i}}) {
                        my $v;
                        while (( undef, $v ) = each %$rpairs) {
                            if (not exists $v->{$field}) {
                                croak("required index field $field ".
                                        "not found");
                            }
                        }
                    }
                    my $key = $makekey->($i);
                    # validate $lock
                    my $lock;
                    my $retry = 0;
                    LOCK:
                    {
                        $lock = $db_pool->{$key}->cds_lock();
                        last LOCK if defined $lock;
                        sleep 3;
                        redo LOCK if ++$retry < $LOCK_RETRY_MAX;
                    }
                    unless (defined $lock) {
                        return ELOCK;
                    }

                    my ( $k, $v, $rc );
                    $rc = 0;
                    while (( $k, $v ) = each %$rpairs) {
                        $rc += $db_pool->{$key}->db_put(
                            $k, freeze($v), DB_NOOVERWRITE);
                    }
                    # no cache in memo
                    $db_pool->{$key}->db_sync();
                    $lock->cds_unlock();
                    return $rc == 0 ? TRUE : EPUT;
                };
            }
            else {
                # Recno by default
                *$put = sub {
                    my ( $self, @hcontent ) = @_;
                    croak("ref $_[1] ne 'HASH'") unless
                      ref $hcontent[0] eq 'HASH';
                    # check required index fields
                    foreach my $field (@{$required_index->{$i}}) {
                        if (not exists $hcontent[0]->{$field}) {
                            croak("required index field $field ".
                                    "not found");
                        }
                    }
                    my @fcontent = map { freeze($_) } @hcontent;
                    my $key = $makekey->($i);
                    # validate $lock
                    my $lock;
                    my $retry = 0;
                    LOCK:
                    {
                        $lock = $db_pool->{$key}->cds_lock();
                        last LOCK if defined $lock;
                        sleep 3;
                        redo LOCK if ++$retry < $LOCK_RETRY_MAX;
                    }
                    unless (defined $lock) {
                        return ELCK;
                    }
                    my ( $k, $first_key, $rc );
                    $rc = 0;
                    foreach (@fcontent) {
                        $k = -1;
                        $rc += $db_pool->{$key}->db_put(
                            $k, $_, DB_APPEND);
                        $first_key = $k unless defined $first_key;
                    }
                    # no cache in memo
                    $db_pool->{$key}->db_sync();
                    $lock->cds_unlock();
                    return $rc == 0 ? $first_key : EPUT;
                };

                *$put2 = sub {
                    my ( $self, $hcontent_array, $key_array ) = @_;
                    return TRUE if @$hcontent_array == 0;
                    croak('ref $_[1]->[0] ne "HASH"') unless
                      ref($hcontent_array->[0]) eq 'HASH';
                    # check required index fields
                    foreach my $field (@{$required_index->{$i}}) {
                        if (not exists $hcontent_array->[0]->{$field}) {
                            croak("required index field $field ".
                                    "not found");
                        }
                    }
                    my $key = $makekey->($i);
                    # validate $lock
                    my $lock;
                    my $retry = 0;
                    LOCK:
                    {
                        $lock = $db_pool->{$key}->cds_lock();
                        last LOCK if defined $lock;
                        sleep 3;
                        redo LOCK if ++$retry < $LOCK_RETRY_MAX;
                    }
                    unless (defined $lock) {
                        return ELOCK;
                    }
                    my ( $k, $rc, $rc_all );
                    $rc_all = 0;
                    foreach my $v (@$hcontent_array) {
                        $k = -1;
                        $rc = $db_pool->{$key}->db_put(
                            $k, freeze($v), DB_APPEND);
                        push @$key_array, $k if $rc == 0;
                        $rc_all += $rc;
                    }
                    # no cache in memo
                    $db_pool->{$key}->db_sync();
                    $lock->cds_unlock();
                    return $rc_all == 0 ? TRUE : EPUT;
                };
            }

            *$upd = sub {
                my ( $self, $k, $hcontent ) = @_;
                croak("ref $_[2] ne 'HASH'") unless
                  ref $hcontent eq 'HASH';
                my ( $v, $rc );
                $v = '';
                my $key = $makekey->($i);
                # validate $lock
                my $lock;
                my $retry = 0;
                LOCK:
                {
                    $lock = $db_pool->{$key}->cds_lock();
                    last LOCK if defined $lock;
                    sleep 3;
                    redo LOCK if ++$retry < $LOCK_RETRY_MAX;
                }
                unless (defined $lock) {
                    return ELCK;
                }
                # FIXME evaluate $cursor
                my $cursor = $db_pool->{$key}->db_cursor(DB_WRITECURSOR);
                $rc = $cursor->c_get($k, $v, DB_SET);
                #print STDERR "key : $k\n";
                #print STDERR "rc  : $rc\n";
                return EUPD unless $rc == 0;
                my $content = thaw($v);
                foreach (keys %$hcontent) {
                    $content->{$_} = $hcontent->{$_};
                }
                my $fcontent = freeze($content);
                $rc = $cursor->c_put($k, $fcontent, DB_CURRENT);
                # no cache in memo
                $db_pool->{$key}->db_sync();
                $lock->cds_unlock();
                $cursor->c_close();
                return $rc == 0 ? 0 : EUPD;
            };

            *$get = sub {
                my ( $self, $k ) = @_;
                my $key = $makekey->($i);
                my ( $fcontent, $rc );
                $rc = $db_pool->{$key}->db_get($k, $fcontent);
                #print STDERR "rc = ", $rc, "\n";
                if ($rc == 0) {
                    return thaw($fcontent);
                }
                else {
                    if ($rc == DB_NOTFOUND or $rc == DB_KEYEMPTY) {
                        return EEPT;
                    }
                    else {
                        if ($DEBUG) {
                            print STDERR __PACKAGE__, ": get_xxx failed\n";
                        }
                        return EGET;
                    }
                }
            };

            *$gets = sub {
                my ( $self, $n, $desc, $offset ) = @_;
                my $key = $makekey->($i);
                my ( $k, $fcontent, $rc );
                $k = -1;
                $desc ||= 0;
                my $cursor = $db_pool->{$key}->db_cursor();
                $rc = $cursor->c_get(
                    $k, $fcontent, $desc ? DB_LAST : DB_FIRST);
                if ($rc == DB_NOTFOUND or $rc == DB_KEYEMPTY) {
                    $cursor->c_close();
                    return [];
                }
                if (defined $offset and $offset > 0) {
                    OFFSET:
                    for (my $i = 0; $rc == 0 or $rc == DB_KEYEMPTY;
                         $rc = $cursor->c_get($k, $fcontent,
                                              $desc ? DB_PREV :
                                                DB_NEXT)
                     ) {
                        last OFFSET if $i == $offset;
                        $i++ if $rc == 0;
                    }
                }
                my $ret = [];
                FETCH:
                for (my $i = 0; $rc == 0 or $rc == DB_KEYEMPTY;
                     $rc = $cursor->c_get(
                         $k, $fcontent, $desc ? DB_PREV : DB_NEXT)) {
                    last FETCH if $i == $n;
                    if ($rc == 0) {
                        my $entry = {
                            KEY     => $k,
                            CONTENT => thaw($fcontent),
                        };
                        push @$ret, $entry;
                        $i++;
                    }
                }
                $cursor->c_close();
                return $ret;
            };

            *$count = sub {
                my $key = $makekey->($i);
                my $stat = $db_pool->{$key}->db_stat();
                if (exists $DB->{$i}->{TYPE} and
                      $DB->{$i}->{TYPE} eq 'Hash') {
                    return $stat->{hash_ndata};
                }
                else {
                    return $stat->{bt_ndata};
                }
            };

            *$dels = sub {
                my ( $self, @n ) = @_;
                return 0 if @n == 0;
                my $key = $makekey->($i);
                my $deleted = 0;
                # validate $lock
                my $lock;
                my $retry = 0;
                LOCK:
                {
                    $lock = $db_pool->{$key}->cds_lock();
                    last LOCK if defined $lock;
                    sleep 3;
                    redo LOCK if ++$retry < $LOCK_RETRY_MAX;
                }
                unless (defined $lock) {
                    return ELCK;
                }
                foreach my $recno (@n) {
                    if ($db_pool->{$key}->db_del($recno) == 0) {
                        $deleted++;
                    }
                }
                $db_pool->{$key}->db_sync();
                $lock->cds_unlock();
                return $deleted;
            };
        }
        else {
            # index database(s)
            foreach my $j (0 .. $#{$DB->{$i}->{SUBS}}) {
                ( my $p = $i ) =~ s/_INDEX$//o;
                my $get = 'get_'. lc($p).
                  's_by_'. lc($DB->{$i}->{SUBS}->[$j]);
                my $cat = 'cat_'. lc($i). '_'. lc($DB->{$i}->{SUBS}->[$j]).
                  's';
                my $count =
                  '__'. lc($i). '_'. lc($DB->{$i}->{SUBS}->[$j]). 's';
                my $countdup =
                  '__'. lc($i). '_'. lc($DB->{$i}->{SUBS}->[$j]). '_dups';

                $STUBS->{$p}->{FIELDS}->{lc($DB->{$i}->{SUBS}->[$j])} =
                  sub { __PACKAGE__->$get(@_); };
                $STUBS->{$i}->{CAT}->{lc($DB->{$i}->{SUBS}->[$j])}
                  = sub { __PACKAGE__->$cat(@_); };
                $STUBS->{$i}->{COUNT}->{lc($DB->{$i}->{SUBS}->[$j])} =
                  sub { __PACKAGE__->$count(@_); };
                $STUBS->{$i}->{COUNTDUP}->{lc($DB->{$i}->{SUBS}->[$j])}
                  = sub { __PACKAGE__->$countdup(@_); };

                *$get = sub {
                    # FIXME ugly api..
                    # TODO hash param
                    my ( $self, $k, $returnValue, $lastone,
                         $n, $offset ) = @_;
                    $returnValue ||= 0;
                    croak("undefined key: $k") unless defined $k;
                    return [] if defined $n and $n <= 0;
                    my $key = $makekey->($i, $DB->{$i}->{SUBS}->[$j]);
                    #print "key = ", $key, "\n";
                    my ( $rc, $pk, $v );
                    my $ret = [];

                    $pk = -1;
                    #$rc = $db_pool->{$key}->db_pget($k, $pk, $v);
                    #if ($rc == DB_NOTFOUND or $rc == DB_KEYEMPTY) {
                    #    return $ret;
                    #}
                    my $cursor = $db_pool->{$key}->db_cursor();
                    #print STDERR "status: ", $db_pool->{$key}->status(),
                    #  "\n";
                    # set cursor to begin of $k slot
                    $rc = $cursor->c_pget($k, $pk, $v, DB_SET);
                    if ($rc == DB_SECONDARY_BAD) {
                        $cursor->c_close();
                        croak("bad secondary index found");
                    }
                    if ($rc == DB_KEYEMPTY or $rc == DB_NOTFOUND) {
                        $cursor->c_close();
                        return $ret;
                    }
                    my $dup_count = 0;
                    $rc = $cursor->c_count($dup_count);
                    unless ($rc == 0) {
                        $cursor->c_close();
                        croak("cannot count duplicate keys");
                    }
                    # offset out of range
                    return $ret if defined $offset and
                      $offset < 0 and -$offset >= $dup_count;

                    if ($n and defined $offset) {
                        # splice
                        if ($offset >= 0) {
                            if ($offset > 0) {
                                OFFSET:
                                for (my $i = 0; $rc == 0;
                                     $rc = $cursor->c_pget(
                                         $k, $pk, $v, DB_NEXT_DUP)) {
                                    last OFFSET if $i == $offset;
                                    $i++;
                                }
                            }
                            my $i = 0;
                            FETCH:
                            {
                                last FETCH if $i == $n;
                                if ($BerkeleyDB::VERSION < 0.29) {
                                # Bug fix for BerkeleyDB v0.27
                                    $pk = unpack("L", $pk) - 1;
                                }
                                if ($returnValue) {
                                    my $entry = {
                                        KEY     => $pk,
                                        CONTENT => thaw($v),
                                    };
                                    push @$ret, $entry;
                                } else {
                                    push @$ret, $pk;
                                }
                                $i++;
                                $pk = -1;
                                redo FETCH if $cursor->c_pget(
                                    $k, $pk, $v, DB_NEXT_DUP) == 0;
                            }
                        }
                        else {
                            # $offset < 0
                            my ( $start_index, $end_index );
                            if ($n >= $dup_count+$offset) {
                                $start_index = 0;
                                $end_index = $dup_count+$offset;
                            }
                            else {
                                $end_index = $dup_count+$offset+1;
                                $start_index = $end_index-$n;
                            }
                            if ($start_index > 0) {
                                OFFSET:
                                for (my $i = 0; $rc == 0;
                                     $rc = $cursor->c_pget(
                                         $k, $pk, $v, DB_NEXT_DUP)) {
                                    last OFFSET if $i++ == $start_index;
                                }
                            }
                            my $i = $start_index;
                            FETCH:
                            {
                                last FETCH if $i > $end_index;
                                if ($BerkeleyDB::VERSION < 0.29) {
                                    # Bug fix for BerkeleyDB v0.27
                                    $pk = unpack("L", $pk) - 1;
                                }
                                if ($returnValue) {
                                    my $entry = {
                                        KEY     => $pk,
                                        CONTENT => thaw($v),
                                    };
                                    push @$ret, $entry;
                                }
                                else {
                                    push @$ret, $pk;
                                }
                                $i++;
                                $pk = -1;
                                redo FETCH if $cursor->c_pget(
                                    $k, $pk, $v, DB_NEXT_DUP) == 0;
                            }
                        }
                        return $ret;
                    }
                    else {
                        my $last;
                        FETCH:
                        {
                            if ($BerkeleyDB::VERSION < 0.29) {
                                # Bug fix for BerkeleyDB v0.27
                                $pk = unpack("L", $pk) - 1;
                            }
                            my $entry;
                            if ($returnValue) {
                                $entry = {
                                    KEY     => $pk,
                                    CONTENT => thaw($v),
                                };
                                push @$ret, $entry;
                            } else {
                                push @$ret, $pk;
                            }
                            $last = $entry;
                            $pk = -1;
                            redo FETCH if $cursor->c_pget(
                                $k, $pk, $v, DB_NEXT_DUP) == 0;
                        }
                        $cursor->c_close();
                        if (not $lastone) {
                            return $ret;
                        } else {
                            return [ $last ];
                        }
                    }
                    # NOREACH
                };

                *$cat = sub {
                    my ( $self, $value ) = @_;
                    my $key = $makekey->($i, $DB->{$i}->{SUBS}->[$j]);
                    #print STDERR
                    #  'database:'. $DB->{$i}->{SUBS}->[$j]. "\n";
                    my $cursor = $db_pool->{$key}->db_cursor();

                    my $ret = {};
                    my ( $k, $pk, $v );
                    $k = $pk = -1;
                    $v = '';
                    # FIXME db_stat first to get key count
                    while ($cursor->c_pget(
                        $k, $pk, $v, DB_NEXT) == 0) {
                        if ($BerkeleyDB::VERSION < 0.29) {
                            # Bug fix for BerkeleyDB v0.27
                            $pk = unpack("L", $pk) - 1;
                            #print STDERR 'length of key:'. length($k). "\n";
                        }
                        if ($value) {
                            push @{$ret->{$k}}, {
                                KEY     => $pk,
                                CONTENT => thaw($v),
                            };
                        }
                        else {
                            push @{$ret->{$k}}, $pk;
                        }
                        $k = $pk = -1;
                        $v = '';
                    }
                    $cursor->c_close();
                    return $ret;
                };

                *$count = sub {
                    my $key = $makekey->($i, $DB->{$i}->{SUBS}->[$j]);
                    my $stat = $db_pool->{$key}->db_stat();
                    return $stat->{bt_ndata};
                };

                *$countdup = sub {
                    my ( $self, $k) = @_;
                    croak("undefined key: $k") unless defined $k;
                    my $key = $makekey->($i, $DB->{$i}->{SUBS}->[$j]);
                    #print "key = ", $key, "\n";
                    my ( $rc, $pk, $v );
                    my $ret;

                    $pk = -1;
                    #$rc = $db_pool->{$key}->db_pget($k, $pk, $v);
                    #if ($rc == DB_NOTFOUND or $rc == DB_KEYEMPTY) {
                    #    return EDUP;
                    #}
                    my $cursor = $db_pool->{$key}->db_cursor();
                    #print STDERR "status: ", $db_pool->{$key}->status(),
                    #  "\n";
                    # set cursor to begin of $k slot
                    $rc = $cursor->c_pget($k, $pk, $v, DB_SET);
                    if ($rc == DB_SECONDARY_BAD) {
                        $cursor->c_close();
                        croak("bad secondary index found");
                    }
                    if ($rc == DB_KEYEMPTY or $rc == DB_NOTFOUND) {
                        $cursor->c_close();
                        return 0;
                    }
                    $ret = EDUP;
                    $rc = $cursor->c_count($ret);
                    $cursor->c_close();
                    return $rc == 0 ? $ret : EDUP;
                };

            }
        }
    }
    my $stubs = '_stubs';
    *$stubs = sub() { $STUBS; };
    use strict 'refs';
    # DIRTY CODE END
}

1;
__END__