File Coverage

blib/lib/Tie/Hash/DBD.pm

Criterion	Covered	Total	%
statement	196	245	80.0
branch	116	154	75.3
condition	13	25	52.0
subroutine	26	46	56.5
pod	2	2	100.0
total	353	472	74.7

package Tie::Hash::DBD;

our $VERSION = "0.26";

use strict;
use warnings;

use Carp;

use DBI;

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

my %DB = (
    # k_asc is needed if h_key mush be converted to hex because
    #       where clause is not permitted on binary/BLOB/...
    Pg		=> {
	temp	=> "temp",
	t_key	=> "bytea primary key",
	t_val	=> "bytea",
	clear	=> "truncate table",
	autoc	=> 0,
	},
    Unify	=> {
	temp	=> "",
	t_key	=> "text",
	t_val	=> "binary",
	clear	=> "delete from",
	k_asc	=> 1,
	},
    Oracle	=> {
	# Oracle does not allow where clauses on BLOB's nor does it allow
	# BLOB's to be primary keys
	temp	=> "global temporary",	# Only as of Ora-9
	t_key	=> "varchar2 (4000) primary key",
	t_val	=> "blob",
	clear	=> "truncate table",
	autoc	=> 0,
	k_asc	=> 1,
	},
    MariaDB	=> {
	temp	=> "temporary",
	t_key	=> "blob",	# Does not allow binary to be primary key
	t_val	=> "blob",
	clear	=> "truncate table",
	autoc	=> 0,
	},
    mysql	=> {
	temp	=> "temporary",
	t_key	=> "blob",	# Does not allow binary to be primary key
	t_val	=> "blob",
	clear	=> "truncate table",
	autoc	=> 0,
	},
    SQLite	=> {
	temp	=> "temporary",
	t_key	=> "text 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	=> "text primary key",
	t_val	=> "text",
	clear	=> "delete from",
	},
    Firebird	=> {
	temp	=> "",
	t_key	=> "varchar (8192)",
	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 TIEHASH {
    my $pkg = shift;
    my $usg = qq{usage: tie %h, "$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,
	asc => $cnf->{k_asc} || 0,
	trh => 0,
	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->{trh} and $h->{trh} = $opt->{trh};
	$opt->{ktp} and $h->{ktp} = $opt->{ktp};
	$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;
    $h->{trh} and $dbh->{AutoCommit} = 0;

    if ($h->{tbl}) {		# Used told the table name
	$dbh->{AutoCommit} = 1 unless $h->{trh} || $dbt eq "CSV" || $dbt eq "Unify";
	}
    else {			# Create a temporary table
	$tmp = ++$dbdx;
	$h->{tbl} = "t_tie_dbdh_$$" . "_$tmp";
	}
    _create_table ($h, $tmp);

    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 = ?");

    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]);
	}

    bless $h, $pkg;
    } # TIEHASH

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 STORE {
    my ($self, $key, $value) = @_;
    $self->{ro} and return _ro_fail ($self, "store entries");
    my $k = $self->{asc} ? unpack "H*", $key : $key;
    my $v = $self->_stream ($value);
    $self->{trh} and $self->{dbh}->begin_work unless $self->{dbt} eq "SQLite";
    my $r = $self->EXISTS ($key)
	? $self->{upd}->execute ($v, $k)
	: $self->{ins}->execute ($k, $v);
    $self->{trh} and $self->{dbh}->commit;
    $r;
    } # STORE

sub DELETE {
    my ($self, $key) = @_;
    $self->{ro} and return _ro_fail ($self, "delete entries");
    $self->{asc} and $key = unpack "H*", $key;
    $self->{trh} and $self->{dbh}->begin_work unless $self->{dbt} eq "SQLite";
    $self->{sel}->execute ($key);
    my $r = $self->{sel}->fetch;
    unless ($r) {
	$self->{trh} and $self->{dbh}->rollback;
	return;
	}

    $self->{del}->execute ($key);
    $self->{trh} and $self->{dbh}->commit;
    $self->_unstream ($r->[0]);
    } # DELETE

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

sub EXISTS {
    my ($self, $key) = @_;
    $self->{asc} and $key = unpack "H*", $key;
    $self->{sel}->execute ($key);
    return $self->{sel}->fetch ? 1 : 0;
    } # EXISTS

sub FETCH {
    my ($self, $key) = @_;
    $self->{asc} and $key = unpack "H*", $key;
    $self->{sel}->execute ($key);
    my $r = $self->{sel}->fetch or return;
    $self->_unstream ($r->[0]);
    } # FETCH

sub FIRSTKEY {
    my $self = shift;
    $self->{trh} and $self->{dbh}->begin_work unless $self->{dbt} eq "SQLite";
    $self->{key} = $self->{dbh}->selectcol_arrayref ("select $self->{f_k} from $self->{tbl}");
    $self->{trh} and $self->{dbh}->commit;
    unless (@{$self->{key}}) {
	$self->{trh} and $self->{dbh}->commit;
	return;
	}
    if ($self->{asc}) {
	 $_ = pack "H*", $_ for @{$self->{key}};
	 }
    pop @{$self->{key}};
    } # FIRSTKEY

sub NEXTKEY {
    my $self = shift;
    unless (@{$self->{key}}) {
	$self->{trh} and $self->{dbh}->commit;
	return;
	}
    pop @{$self->{key}};
    } # FIRSTKEY

sub SCALAR {
    my $self = shift;
    $self->{cnt}->execute;
    my $r = $self->{cnt}->fetch or return 0;
    $r->[0];
    } # SCALAR

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

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

sub DESTROY {
    my $self = shift;
    my $dbh = $self->{dbh} or return;
    for (qw( sel ins upd del cnt ctv )) {
	$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__