File Coverage

blib/lib/Tie/Array/DBD.pm

Criterion	Covered	Total	%
statement	238	301	79.0
branch	126	170	74.1
condition	10	17	58.8
subroutine	31	55	56.3
pod	2	2	100.0
total	407	545	74.6

package Tie::Array::DBD;

our $VERSION = "0.26";

use strict;
use warnings;

use Carp;

use DBI;

my $dbdx = sprintf "%04d", (time + int rand 10000) % 10000;

my %DB = (
    Pg		=> {
	temp	=> "temp",
	t_key	=> "bigint not null primary key",
	t_val	=> "bytea",
	clear	=> "truncate table",
	autoc	=> 0,
	},
    Unify	=> {
	temp	=> "",
	t_key	=> "numeric (9) not null primary key",
	t_val	=> "binary",
	clear	=> "delete from",
	},
    Oracle	=> {
	temp	=> "global temporary",	# Only as of Ora-9
	t_key	=> "number (38) not null primary key",
	t_val	=> "blob",
	clear	=> "truncate table",
	autoc	=> 0,
	},
    MariaDB	=> {
	temp	=> "temporary",
	t_key	=> "bigint not null primary key",
	t_val	=> "blob",
	clear	=> "truncate table",
	autoc	=> 0,
	},
    mysql	=> {
	temp	=> "temporary",
	t_key	=> "bigint not null primary key",
	t_val	=> "blob",
	clear	=> "truncate table",
	autoc	=> 0,
	},
    SQLite	=> {
	temp	=> "temporary",
	t_key	=> "integer not null primary key",
	t_val	=> "blob",
	clear	=> "delete from",
	pbind	=> 0, # TYPEs in SQLite are text, bind_param () needs int
	autoc	=> 0,
	},
    CSV		=> {
	temp	=> "temporary",
	t_key	=> "integer not null primary key",
	t_val	=> "text",
	clear	=> "delete from",
	},
    Firebird	=> {
	temp	=> "",
	t_key	=> "integer primary key",
	t_val	=> "varchar (8192)",
	clear	=> "delete from",
	},
    );

sub _ro_fail {
    my ($self, $action) = @_;
    my $error = "You cannot $action when in read-only mode";
    $self->{ro} == 1 ? warn $error : die $error;
    } # ro_fail

sub _create_table {
    my ($cnf, $tmp) = @_;
    $cnf->{tmp} = $tmp;

    my $dbh = $cnf->{dbh};
    my $dbt = $cnf->{dbt};

    my $exists = 0;
    eval {
	local $dbh->{PrintError} = 0;
	my $sth = $dbh->prepare ("select $cnf->{f_k}, $cnf->{f_v} from $cnf->{tbl}");
	$sth->execute;
	$cnf->{tmp} = 0;
	$exists = 1;
	};
    $exists and return;	# Table already exists

    $cnf->{ro} and return _ro_fail ($cnf, "create tables");
    my $temp = $DB{$dbt}{temp};
    $cnf->{tmp} or $temp = "";
    local $dbh->{AutoCommit} = 1 unless $dbt eq "CSV" || $dbt eq "Unify";
    $dbh->do (
	"create $temp table $cnf->{tbl} (".
	    "$cnf->{f_k} $cnf->{ktp},".
	    "$cnf->{f_v} $cnf->{vtp})"
	);
    $dbt eq "Unify" and $dbh->commit;
    } # create table

sub TIEARRAY {
    my $pkg = shift;
    my $usg = qq{usage: tie \@a, "$pkg", \$dbh [, { tbl => "tbl", key => "f_key", fld => "f_value" }];};
    my $dsn = shift or croak $usg;
    my $opt = shift;

    my $dbh = ref $dsn
	? $dsn->clone
	: DBI->connect ($dsn, undef, undef, {
	    PrintError       => 1,
	    RaiseError       => 1,
	    PrintWarn        => 0,
	    FetchHashKeyName => "NAME_lc",
	    }) or croak DBI->errstr;

    my $dbt = $dbh->{Driver}{Name} || "no DBI handle";
    my $cnf = $DB{$dbt} or croak "I don't support database '$dbt'";
    my $f_k = "h_key";
    my $f_v = "h_value";
    my $tmp = 0;

    $dbh->{PrintWarn}   = 0;
    $dbh->{AutoCommit}  = $cnf->{autoc} if exists $cnf->{autoc};
    $dbh->{LongReadLen} = 4_194_304     if $dbt eq "Oracle";

    my $h = {
	dbt => $dbt,
	dbh => $dbh,
	tbl => undef,
	tmp => $tmp,
	ktp => $cnf->{t_key},
	vtp => $cnf->{t_val},
	ro  => 0,

	_en => undef,
	_de => undef,
	};

    if ($opt) {
	ref $opt eq "HASH" or croak $usg;

	$opt->{key} and $f_k      = $opt->{key};
	$opt->{fld} and $f_v      = $opt->{fld};
	$opt->{tbl} and $h->{tbl} = $opt->{tbl};
	$opt->{vtp} and $h->{vtp} = $opt->{vtp};

	exists $opt->{ro} and $h->{ro} = $opt->{ro};

	if (my $str = $opt->{str}) {
	    if ($str eq "Sereal") {
		require Sereal::Encoder;
		require Sereal::Decoder;
		my $se = Sereal::Encoder->new;
		my $sd = Sereal::Decoder->new;
		$h->{_en} = sub { $se->encode ($_[0]) };
		$h->{_de} = sub { $sd->decode ($_[0]) };
		}
	    elsif ($str eq "Storable") {
		require Storable;
		$h->{_en} = sub { Storable::nfreeze ({ val => $_[0] }) };
		$h->{_de} = sub { Storable::thaw    ($_[0])->{val}     };
		}
	    elsif ($str eq "FreezeThaw") {
		require FreezeThaw;
		$h->{_en} = sub {  FreezeThaw::freeze ($_[0])     };
		$h->{_de} = sub { (FreezeThaw::thaw   ($_[0]))[0] };
		}
	    elsif ($str eq "JSON") {
		require JSON;
		my $j = JSON->new->allow_nonref;
		$h->{_en} = sub { $j->utf8->encode ($_[0]) };
		$h->{_de} = sub {       $j->decode ($_[0]) };
		}
	    elsif ($str eq "JSON::MaybeXS") {
		require JSON::MaybeXS;
		my $j = JSON::MaybeXS->new->allow_nonref;
		$h->{_en} = sub { $j->utf8->encode ($_[0]) };
		$h->{_de} = sub {       $j->decode ($_[0]) };
		}
	    elsif ($str eq "JSON::SIMD") {
		require JSON::SIMD;
		my $j = JSON::SIMD->new->allow_nonref;
		$h->{_en} = sub { $j->utf8->encode ($_[0]) };
		$h->{_de} = sub {       $j->decode ($_[0]) };
		}
	    elsif ($str eq "JSON::Syck") {
		require JSON::Syck;
		$h->{_en} = sub { JSON::Syck::Dump ($_[0]) };
		$h->{_de} = sub { JSON::Syck::Load ($_[0]) };
		}
	    elsif ($str eq "JSON::XS") {
		require JSON::XS;
		my $j = JSON::XS->new->allow_nonref;
		$h->{_en} = sub { $j->utf8->encode ($_[0]) };
		$h->{_de} = sub {       $j->decode ($_[0]) };
		}
	    elsif ($str eq "YAML") {
		require YAML;
		$h->{_en} = sub { YAML::Dump ($_[0]) };
		$h->{_de} = sub { YAML::Load ($_[0]) };
		}
	    elsif ($str eq "YAML::Syck") {
		require YAML::Syck;
		$h->{_en} = sub { YAML::Syck::Dump ($_[0]) };
		$h->{_de} = sub { YAML::Syck::Load ($_[0]) };
		}
	    elsif ($str eq "Data::Dumper") {
		require Data::Dumper;
		$h->{_en} = sub { Data::Dumper::Dumper ($_[0]) };
		$h->{_de} = sub { eval $_[0] };
		}
	    elsif ($str eq "XML::Dumper") {
		require XML::Dumper;
		my $xd = XML::Dumper->new;
		$h->{_en} = sub { $xd->pl2xml ($_[0]) };
		$h->{_de} = sub { $xd->xml2pl ($_[0]) };
		}
	    elsif ($str eq "Bencode") {
		require Bencode;
		$h->{_en} = sub { Bencode::bencode ($_[0]) };
		$h->{_de} = sub { Bencode::bdecode ($_[0]) };
		}
	    else {
		croak "Unsupported serializer: $str\n";
		}
	    }
	}

    $h->{f_k} = $f_k;
    $h->{f_v} = $f_v;

    unless ($h->{tbl}) {	# Create a temporary table
	$tmp = ++$dbdx;
	$h->{tbl} = "t_tie_dbda_$$" . "_$tmp";
	}
    _create_table ($h, $tmp);
    _setmax ($h);

    my $tbl = $h->{tbl};

    $h->{ins} = $dbh->prepare ("insert into $tbl ($f_k, $f_v) values (?, ?)") unless $h->{ro};
    $h->{del} = $dbh->prepare ("delete from $tbl where $f_k = ?")             unless $h->{ro};
    $h->{upd} = $dbh->prepare ("update $tbl set $f_v = ? where $f_k = ?")     unless $h->{ro};
    $h->{sel} = $dbh->prepare ("select $f_v from $tbl where $f_k = ?");
    $h->{cnt} = $dbh->prepare ("select count(*) from $tbl");
    $h->{ctv} = $dbh->prepare ("select count(*) from $tbl where $f_k = ?");
    $h->{uky} = $dbh->prepare ("update $tbl set $f_k = ? where $f_k = ?")     unless $h->{ro};

    unless (exists $cnf->{pbind} && !$cnf->{pbind}) {
	my $sth = $dbh->prepare ("select $f_k, $f_v from $tbl where 0 = 1");
	$sth->execute;
	my @typ = @{$sth->{TYPE}};

	$h->{ins}->bind_param (1, undef, $typ[0]);
	$h->{ins}->bind_param (2, undef, $typ[1]);
	$h->{del}->bind_param (1, undef, $typ[0]);
	$h->{upd}->bind_param (1, undef, $typ[1]);
	$h->{upd}->bind_param (2, undef, $typ[0]);
	$h->{sel}->bind_param (1, undef, $typ[0]);
	$h->{ctv}->bind_param (1, undef, $typ[0]);
	$h->{uky}->bind_param (1, undef, $typ[0]);
	$h->{uky}->bind_param (2, undef, $typ[0]);
	}

    bless $h, $pkg;
    } # TIEARRAY

sub _stream {
    my ($self, $val) = @_;
    defined $val or return undef;

    $self->{_en} and return $self->{_en}->($val);
    return $val;
    } # _stream

sub _unstream {
    my ($self, $val) = @_;
    defined $val or return undef;

    $self->{_de} and return $self->{_de}->($val);
    return $val;
    } # _unstream

sub _setmax {
    my $self = shift;
    my $sth = $self->{dbh}->prepare ("select max($self->{f_k}) from $self->{tbl}");
    $sth->execute;
    if (my $r = $sth->fetch) {
	$self->{max} = defined $r->[0] ? $r->[0] : -1;
	}
    else {
	$self->{max} = -1;
	}
    $self->{max};
    } # _setmax

sub STORE {
    my ($self, $key, $value) = @_;
    $self->{ro} and return _ro_fail ($self, "store entries");
    my $v = $self->_stream ($value);
    $self->EXISTS ($key)
	? $self->{upd}->execute ($v, $key)
	: $self->{ins}->execute ($key, $v);
    $key > $self->{max} and $self->{max} = $key;
    } # STORE

sub DELETE {
    my ($self, $key) = @_;
    $self->{ro} and return _ro_fail ($self, "delete entries");
    $self->{sel}->execute ($key);
    my $r = $self->{sel}->fetch or return;
    $self->{del}->execute ($key);
    $key >= $self->{max} and $self->_setmax;
    $self->_unstream ($r->[0]);
    } # DELETE

sub STORESIZE {
    my ($self, $size) = @_; # $size = $# + 1
    $self->{ro} and return _ro_fail ($self, "resize an array");
    $size--;
    $self->{dbh}->do ("delete from $self->{tbl} where $self->{f_k} > $size");
    $self->{max} = $size;
    } # STORESIZE

sub CLEAR {
    my $self = shift;
    $self->{ro} and return _ro_fail ($self, "clear an array");
    $self->{dbh}->do ("$DB{$self->{dbt}}{clear} $self->{tbl}");
    $self->{max} = -1;
    } # CLEAR

sub EXISTS {
    my ($self, $key) = @_;
    $key <= $self->{max} or return 0;
    $self->{sel}->execute ($key);
    return $self->{sel}->fetch ? 1 : 0;
    } # EXISTS

sub FETCH {
    my ($self, $key) = @_;
    $key <= $self->{max} or return undef;
    $self->{sel}->execute ($key);
    my $r = $self->{sel}->fetch or return;
    $self->_unstream ($r->[0]);
    } # STORE

sub PUSH {
    my ($self, @val) = @_;
    $self->{ro} and return _ro_fail ($self, "push entries");
    $self->STORE (++$self->{max}, $_) for @val;
    return $self->FETCHSIZE;
    } # PUSH

sub POP {
    my $self = shift;
    $self->{ro} and return _ro_fail ($self, "pop entries");
    $self->{max} >= 0 or return;
    $self->DELETE ($self->{max});
    } # POP

sub SHIFT {
    my $self = shift;
    $self->{ro} and return _ro_fail ($self, "shift entries");
    my $val  = $self->DELETE (0);
    $self->{uky}->execute ($_ - 1, $_) for 1 .. $self->{max};
    $self->{max}--;
    return $val;
    } # SHIFT

sub UNSHIFT {
    my ($self, @val) = @_;
    @val or return;
    $self->{ro} and return _ro_fail ($self, "unshift entries");
    my $incr = scalar @val;
    $self->{uky}->execute ($_ + $incr, $_) for reverse 0 .. $self->{max};
    $self->{max} += $incr;
    $self->STORE ($_, $val[$_]) for 0 .. $#val;
    return $self->FETCHSIZE;
    } # UNSHIFT

# splice ARRAY, OFFSET, LENGTH, LIST
# splice ARRAY, OFFSET, LENGTH
# splice ARRAY, OFFSET
# splice ARRAY
#
#   Removes the elements designated by OFFSET and LENGTH from an array, and
#   replaces them with the elements of LIST, if any.
#
#   In list   context, returns the elements removed from the array.
#   In scalar context, returns the last element removed, or "undef" if
#    no elements are removed.
#
#   The array grows or shrinks as necessary.
#
#   If OFFSET is negative then it starts that far from the end of the array.
#   If LENGTH is omitted, removes everything from OFFSET onward.
#   If LENGTH is negative, removes the elements from OFFSET onward except for
#     -LENGTH elements at the end of the array.
#   If both OFFSET and LENGTH are omitted, removes everything.
#   If OFFSET is past the end of the array, Perl issues a warning, and splices
#     at the end of the array.

sub SPLICE {
    my $nargs = $#_;
    my ($self, $off, $len, @new, @val) = @_;

    $self->{ro} and return _ro_fail ($self, "splice entries");
    # splice @array;
    if ($nargs == 0) {
	if (wantarray) {
	    @val = map { $self->FETCH ($_) } 0 .. $self->{max};
	    $self->CLEAR;
	    return @val;
	    }
	$val[0] = $self->FETCH ($self->{max});
	$self->CLEAR;
	return $val[0];
	}

    # Take care of negative offset, count from tail
    $off < 0 and $off = $self->{max} + 1 + $off;
    $off < 0 and
	croak "Modification of non-creatable array value attempted, subscript $_[1]";

    # splice @array, off;
    if ($nargs == 1) {
	$off > $self->{max} and return;

	if (wantarray) {
	    @val = map { $self->FETCH ($_) } $off .. $self->{max};
	    $self->STORESIZE ($off);
	    return @val;
	    }
	$val[0] = $self->FETCH ($self->{max});
	$self->STORESIZE ($off);
	return $val[0];
	}

    # splice @array, off, len;
    $nargs == 2 && $off  > $self->{max} and return;

    my $last = $len < 0 ? $self->{max} + $len : $off + $len - 1;
    $nargs == 2 && $last > $self->{max} and return $self->SPLICE ($off);

    @val = map { $self->DELETE ($_) } $off .. $last;
    $len = @val;
    $self->{uky}->execute ($_ - $len, $_) for ($last + 1) .. $self->{max};
    $self->{max} -= $len;

    # splice @array, off, len, replacement-list;
    if (@new) {
	my $new = @new;
	$self->{uky}->execute ($_ + $new, $_) for reverse $off .. $self->{max};
	$self->STORE ($off + $_, $new[$_]) for 0..$#new;
	$self->{max} += $new;
	}

    return wantarray ? @val : $val[-1];
    } # SPLICE

sub FIRSTKEY {
    my $self = shift;
    $self->{max} >= 0 or return;
    $self->{min} = 0;
    } # FIRSTKEY

sub NEXTKEY {
    my $self = shift;
    exists $self->{min} && $self->{min} < $self->{max} and return ++$self->{min};
    delete $self->{min};
    return;
    } # FIRSTKEY

sub FETCHSIZE {
    my $self = shift;
    return $self->{max} + 1;
    } # FETCHSIZE

sub EXTEND {
    # no-op
    } # EXTEND

sub drop {
    my $self = shift;
    $self->{tmp} = 1;
    } # drop

sub readonly {
    my $self = shift;
    @_ and $self->{ro} = shift;
    return $self->{ro};
    } # readonly

sub _dump_table {
    my $self = shift;
    my $sth = $self->{dbh}->prepare ("select $self->{f_k}, $self->{f_v} from $self->{tbl} order by $self->{f_k}");
    $sth->execute;
    $sth->bind_columns (\my ($k, $v));
    while ($sth->fetch) {
	printf STDERR "%6d: '%s'\n", $k, $self->_unstream ($v);
	}
    } # _dump_table

sub DESTROY {
    my $self = shift;
    my $dbh = $self->{dbh} or return;
    for (qw( sel ins upd del cnt ctv uky )) {
	$self->{$_} or next;
	$self->{$_}->finish;
	undef  $self->{$_}; # DESTROY handle
	delete $self->{$_};
	}
    delete $self->{$_} for qw( _de _en );
    if ($self->{tmp}) {
	$dbh->{AutoCommit} or $dbh->rollback;
	$self->{ro} and return _ro_fail ($self, "drop tables");
	$dbh->do ("drop table ".$self->{tbl});
	}
    $dbh->{AutoCommit} or $dbh->commit;
    $dbh->disconnect;
    undef $dbh;
    undef $self->{dbh};
    } # DESTROY

1;

__END__