You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1353 lines
42 KiB
1353 lines
42 KiB
4 months ago
|
#
|
||
|
|
# $Id: dm.pm,v 1.1.1.1 2004/01/09 09:22:07 jeffo Exp $
|
||
|
|
#
|
||
|
|
# $Source: /usr/local/cvsroot/Testmanager/cgi-bin/dm.pm,v $
|
||
|
|
#
|
||
|
|
# Routines for manipulating (ie., querying, updating) a data model.
|
||
|
|
#
|
||
|
|
#############################################################################
|
||
|
|
# #
|
||
|
|
# COPYRIGHT NOTICE #
|
||
|
|
# #
|
||
|
|
# This code was written by Ed Loehr, all rights reserved, 1999. You are #
|
||
|
|
# hereby granted a non-exclusive license to do whatever you want with #
|
||
|
|
# this code, including modification and redistribution, so long as you #
|
||
|
|
# retain this copyright notice in the code and do not attempt to prevent #
|
||
|
|
# anyone else from the same privileges. #
|
||
|
|
# #
|
||
|
|
# ALL SOFTWARE HAS BUGS. THIS CODE COMES "AS-IS" WITH ABSOLUTELY NO #
|
||
|
|
# WARRANTY OR GUARANTEE OF ITS FITNESS FOR ANY PURPOSE WHATSOEVER. #
|
||
|
|
# BY INCORPORATING THIS CODE, YOU ASSUME COMPLETE RISK FOR ANY #
|
||
|
|
# CONSEQUENCES AND AGREE NOT TO BRING ANY GRIEVANCE OR LIABILITTY #
|
||
|
|
# CLAIMS WHATSOVER AGAINST Ed Loehr. #
|
||
|
|
# #
|
||
|
|
#############################################################################
|
||
|
|
|
||
|
|
package dm;
|
||
|
|
|
||
|
|
use DBI;
|
||
|
|
use logger;
|
||
|
|
use toolbox;
|
||
|
|
|
||
|
|
@dm::initOptions = qw( dbkind dbhost dbname dbport dbuser dbpasswd dbhname );
|
||
|
|
BEGIN {
|
||
|
|
$dm::default_dbhname = 'tmdbh';
|
||
|
|
$dm::dbDefault{paramsep} = ' | ';
|
||
|
|
$dm::dbDefault{dbkind} = "Pg";
|
||
|
|
$dm::dbDefault{dbhost} = $ENV{PGHOST} || "localhost";
|
||
|
|
$dm::dbDefault{dbname} = $ENV{PGDATABASE} || `whoami`;
|
||
|
|
chomp $dm::dbDefault{dbname};
|
||
|
|
$dm::dbDefault{dbport} = $ENV{PGPORT} || 5432;
|
||
|
|
$dm::dbDefault{dbuser} = $ENV{PGUSER} || `whoami`;
|
||
|
|
chomp $dm::dbDefault{dbuser};
|
||
|
|
$dm::dbDefault{dbpasswd} = undef;
|
||
|
|
$dm::dbDefault{dbhname} = $dm::default_dbhname;
|
||
|
|
}
|
||
|
|
|
||
|
|
$dm::app_abbreviation = 'tm';
|
||
|
|
$dm::run_dir = "/tmp/".$dm::app_abbreviation."/run";
|
||
|
|
$dm::down_file = "$dm::run_dir/dbdown_$dm::app_abbreviation";
|
||
|
|
$dm::touchfile_dir = $dm::run_dir;
|
||
|
|
$dm::log_touchfile = "$dm::touchfile_dir/log_$dm::app_abbreviation";
|
||
|
|
$dm::debug_touchfile = "$dm::touchfile_dir/debug_$dm::app_abbreviation";
|
||
|
|
$dm::profile_touchfile = "$dm::touchfile_dir/profile_$dm::app_abbreviation";
|
||
|
|
$dm::explain_touchfile = "$dm::touchfile_dir/explain_$dm::app_abbreviation";
|
||
|
|
$dm::showdata_touchfile = "$dm::touchfile_dir/showdata_$dm::app_abbreviation";
|
||
|
|
|
||
|
|
$dm::cleanup_interval = 100; # cleanup (vacuum) after this many transactions...
|
||
|
|
$dm::query_count = 0;
|
||
|
|
$dm::reconnect_tries = 0;
|
||
|
|
$dm::reconnect_limit = 2;
|
||
|
|
$dm::vacuum_tries = 0;
|
||
|
|
$dm::vacuum_limit = 1;
|
||
|
|
|
||
|
|
#############################################################################
|
||
|
|
## Query Cache NOTES... ############################################
|
||
|
|
#############################################################################
|
||
|
|
# This is a poor-man's query cache. We k=
|
||
|
|
#############################################################################
|
||
|
|
## Query Cache Settings... ############################################
|
||
|
|
#############################################################################
|
||
|
|
# FIXME: Turning the query cache off for now because the race condition
|
||
|
|
# FIXME: is too nasty in the problems it leads to. I think we really must
|
||
|
|
# FIXME: push any query caching of dynamic data down into the DB.
|
||
|
|
$dm::query_cache_enabled = 1;
|
||
|
|
$dm::log_cache_actions = 0;
|
||
|
|
$dm::log_cache_savings = 0;
|
||
|
|
$dm::cache_everything = 0;
|
||
|
|
# I dropped the supporting triggers for this table dependency
|
||
|
|
# functionality. They must be present for this to work!!! -efl, May2001
|
||
|
|
$dm::use_dependent_table_strategy = 0;
|
||
|
|
$dm::derive_unstated_table_dependencies = 0;
|
||
|
|
%dm::query_cache = ();
|
||
|
|
$dm::query_cache_size = 0;
|
||
|
|
$dm::query_cache_hits = 0;
|
||
|
|
$dm::query_cache_misses = 0;
|
||
|
|
$dm::show_query_cache_hitmiss = 1;
|
||
|
|
$dm::max_cache_size = 2000000;
|
||
|
|
$dm::last_table_status_check_time = 1;
|
||
|
|
%dm::table_status = (); # storage of last update time for tables ...
|
||
|
|
$dm::tolerable_cache_race_window = 0.51; # seconds
|
||
|
|
#############################################################################
|
||
|
|
|
||
|
|
$dm::pending_queries = 0;
|
||
|
|
$dm::dberrstr = undef; # For forwarding to client upon request
|
||
|
|
|
||
|
|
#
|
||
|
|
# Postgres doesn't return error codes, only msgs, so we have to handle them...
|
||
|
|
#
|
||
|
|
$dm::PGIDX_SCAN_ERR = "ExecInitIndexScan";
|
||
|
|
$dm::PGIDX_SCAN_ERR_LEN = length($dm::PGIDX_SCAN_ERR);
|
||
|
|
|
||
|
|
#
|
||
|
|
use constant PGSQL => 1;
|
||
|
|
use constant ORACLE => 2;
|
||
|
|
use constant SQLFLEX => 3;
|
||
|
|
use constant SYBASE => 4;
|
||
|
|
|
||
|
|
$dm::dbtype = dm::PGSQL;
|
||
|
|
|
||
|
|
$dm::time_to_die = 0;
|
||
|
|
sub TimeToDie {
|
||
|
|
return $dm::time_to_die;
|
||
|
|
}
|
||
|
|
|
||
|
|
sub SetTimeToDie {
|
||
|
|
$dm::time_to_die = 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
sub SetCacheRaceTolerance {
|
||
|
|
my ($tol) = shift;
|
||
|
|
$dm::tolerable_cache_race_window = (defined($tol) ? $tol : 0.26);
|
||
|
|
return $tol;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub getDownFileName {
|
||
|
|
return $dm::down_file;
|
||
|
|
}
|
||
|
|
|
||
|
|
sub systemDown {
|
||
|
|
if ( ! -f getDownFileName() ) {
|
||
|
|
return 0;
|
||
|
|
}
|
||
|
|
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
sub getDBErrStr {
|
||
|
|
return $dm::dberrstr;
|
||
|
|
}
|
||
|
|
|
||
|
|
sub getRawDBErrStr {
|
||
|
|
return $dm::rawdberrstr;
|
||
|
|
}
|
||
|
|
|
||
|
|
sub setRawDBErrStr {
|
||
|
|
my ($str) = @_;
|
||
|
|
if ( defined($str) && length($str) ) {
|
||
|
|
$dm::rawdberrstr = $str;
|
||
|
|
} else {
|
||
|
|
$dm::rawdberrstr = undef;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
sub setDBErrStr {
|
||
|
|
my ($str) = @_;
|
||
|
|
if ( defined($str) && length($str) ) {
|
||
|
|
setRawDBErrStr($str);
|
||
|
|
$dm::dberrstr = logger::timestamp()." $str";
|
||
|
|
} else {
|
||
|
|
$dm::dberrstr = undef;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
sub addRawDBErrStr {
|
||
|
|
my ($str) = @_;
|
||
|
|
if ( defined($str) && length($str) ) {
|
||
|
|
$dm::rawdberrstr .= $str;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
sub addDBErrStr {
|
||
|
|
my ($str) = @_;
|
||
|
|
if ( defined($str) && length($str) ) {
|
||
|
|
addRawDBErrStr($str);
|
||
|
|
$dm::dberrstr = logger::timestamp()." $str";
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
#
|
||
|
|
# I wanted a really easy way to flip on debugging during development,
|
||
|
|
# so I made this rather crude file system switch that allows me to
|
||
|
|
# turn on debugging output without restarting any systems. This may
|
||
|
|
# not be too desirable in a hi-perf situation, but it's worked well
|
||
|
|
# so far for my development purposes.
|
||
|
|
#
|
||
|
|
sub SetDBGSettings {
|
||
|
|
my ($opts) = @_;
|
||
|
|
if ( -f $dm::log_touchfile || $opts->{log} ) {
|
||
|
|
$dm::log = 1;
|
||
|
|
} else {
|
||
|
|
$dm::log = 0;
|
||
|
|
}
|
||
|
|
if ( -f $dm::debug_touchfile || $opts->{debug} ) {
|
||
|
|
$dm::debug = 1;
|
||
|
|
} else {
|
||
|
|
$dm::debug = 0;
|
||
|
|
}
|
||
|
|
if ( -f $dm::showdata_touchfile || $opts->{showdata} ) {
|
||
|
|
$dm::showdata = 1;
|
||
|
|
} else {
|
||
|
|
$dm::showdata = 0;
|
||
|
|
}
|
||
|
|
if ( -f $dm::profile_touchfile || $opts->{profile} ) {
|
||
|
|
$dm::profile = 1;
|
||
|
|
} else {
|
||
|
|
$dm::profile = 0;
|
||
|
|
}
|
||
|
|
if ( -f $dm::explain_touchfile || $opts->{explain} ) {
|
||
|
|
$dm::explain = 1;
|
||
|
|
} else {
|
||
|
|
$dm::explain = 0;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
sub GetDBH {
|
||
|
|
my ($opts) = @_;
|
||
|
|
|
||
|
|
my $dbh;
|
||
|
|
eval { $dbh = _GetDBH($opts); };
|
||
|
|
if ( $@ ) {
|
||
|
|
logger::logerr($@);
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
return $dbh;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub _GetDBH {
|
||
|
|
my ($opts) = @_;
|
||
|
|
|
||
|
|
|
||
|
|
foreach my $o ( @initOptions ) {
|
||
|
|
$dm::cfg{$o} = ($opts->{$o} ||= $dm::dbDefault{$o});
|
||
|
|
$opts->{$o} = $dm::cfg{$o} if ( $opts->{loadme} );
|
||
|
|
logger::loginfo("dm::cfg{$o} = '$dm::cfg{$o}'") if ( $dm::debug );
|
||
|
|
}
|
||
|
|
|
||
|
|
my $options = "";
|
||
|
|
my $dsn = "dbi:$dm::cfg{dbkind}:dbname=$dm::cfg{dbname};host=$dm::cfg{dbhost};port=$dm::cfg{dbport}";
|
||
|
|
$dm::cfg{dbopts} = {
|
||
|
|
ChopBlanks => defined($dm::cfg{ChopBlanks}) ? $dm::cfg{ChopBlanks} : 1,
|
||
|
|
AutoCommit => defined($dm::cfg{AutoCommit}) ? $dm::cfg{AutoCommit} : 0,
|
||
|
|
};
|
||
|
|
|
||
|
|
logger::loginfo("Retrieving DBH for '$dm::cfg{dbhname}' (default '$dm::dbDefault{dbhname}')...") if ( $dm::debug );
|
||
|
|
# $dm::dbhCache; # hash of various db handles...
|
||
|
|
if ( ! defined($dm::dbhCache{$dm::cfg{dbhname}}) ) {
|
||
|
|
logger::loginfo("Connecting to $dsn as user $dm::cfg{dbuser}") if ( $dm::log );
|
||
|
|
|
||
|
|
my $t0 = toolbox::TimeNow();
|
||
|
|
logger::loginfo("DBI->connect($dsn, $dm::cfg{dbuser}, $dm::cfg{dbpasswd})") if ( $dm::log );
|
||
|
|
$dm::dbhCache{$dm::cfg{dbhname}} = DBI->connect($dsn, $dm::cfg{dbuser}, $dm::cfg{dbpasswd}, $dm::cfg{dbopts});
|
||
|
|
if ( ! $dm::dbhCache{$dm::cfg{dbhname}} ) {
|
||
|
|
logger::logerr("DBI->connect($dsn, $dm::cfg{dbuser}, $dm::cfg{dbpasswd}, { ".join($dm::cfg{paramsep}, $dm::cfg{dbopts})." }) FAILED; DBI::errstr = [$DBI::errstr]");
|
||
|
|
} elsif ($dm::profile) {
|
||
|
|
logger::loginfo("Successfully connected to database $dsn");
|
||
|
|
logger::loginfo(toolbox::ElapsedTimeString("DBI connect: %.3f secs\n",$t0));
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
return $dm::dbhCache{$dm::cfg{dbhname}};
|
||
|
|
}
|
||
|
|
|
||
|
|
sub CloseDB {
|
||
|
|
my ($opts) = @_;
|
||
|
|
my $dbhname = $opts->{dbhname} || $dm::cfg{dbhname};
|
||
|
|
if ( defined($dm::dbhCache{$dbhname}) ) {
|
||
|
|
logger::loginfo("Nuking db handle for $dbhname ...") if ( $dm::log );
|
||
|
|
$dm::dbhCache{$dbhname}->disconnect;
|
||
|
|
$dm::dbhCache{$dbhname} = undef;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
SetDBGSettings;
|
||
|
|
|
||
|
|
sub QueryCount( $ ) {
|
||
|
|
my $reset = shift;
|
||
|
|
my $count = $dm::query_count;
|
||
|
|
if ( $reset ) {
|
||
|
|
$dm::query_count = 0;
|
||
|
|
%dm::query_profile = ();
|
||
|
|
}
|
||
|
|
|
||
|
|
return $count;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub IncQueryCount {
|
||
|
|
my ($sql,$et) = @_;
|
||
|
|
$dm::query_profile{$sql}{'count'}++;
|
||
|
|
$dm::query_profile{$sql}{'et'} += $et;
|
||
|
|
return ( ++$dm::query_count );
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
sub DumpQueryProfile {
|
||
|
|
#
|
||
|
|
# Sort first by total time spent in the query, then by avg time
|
||
|
|
# per query...
|
||
|
|
#
|
||
|
|
my $sort_func = sub {
|
||
|
|
( $dm::query_profile{$a}{'et'} <=>
|
||
|
|
$dm::query_profile{$b}{'et'} ) ||
|
||
|
|
# ( ($dm::query_profile{$a}{'et'} /
|
||
|
|
# $dm::query_profile{$a}{'count'}) <=>
|
||
|
|
# ($dm::query_profile{$b}{'et'} /
|
||
|
|
# $dm::query_profile{$b}{'count'}) ) ||
|
||
|
|
($dm::query_profile{$a}{'count'} <=>
|
||
|
|
$dm::query_profile{$b}{'count'})
|
||
|
|
};
|
||
|
|
|
||
|
|
foreach my $q ( sort $sort_func keys %dm::query_profile ) {
|
||
|
|
print STDERR "##############################################\n";
|
||
|
|
print STDERR "##############################################\n";
|
||
|
|
print STDERR "##############################################\n";
|
||
|
|
print STDERR "QUERY: [$q]\n";
|
||
|
|
printf STDERR sprintf("TOT ET: %.3f secs\n",
|
||
|
|
$dm::query_profile{$q}{'et'});
|
||
|
|
print STDERR "COUNT: [".$dm::query_profile{$q}{'count'}."]\n";
|
||
|
|
my $avg = ($dm::query_profile{$q}{'count'} ?
|
||
|
|
($dm::query_profile{$q}{'et'} /
|
||
|
|
$dm::query_profile{$q}{'count'}) : 0);
|
||
|
|
printf STDERR sprintf("AVG ET: %.3f secs\n", $avg);
|
||
|
|
print STDERR "CACHE HITS: ".($dm::query_profile{$q}{'hits'}||0)."\n";
|
||
|
|
print STDERR "CACHE SAVINGS: ".sprintf("%.3f secs\n", (($dm::query_profile{$q}{'hits'}||0)*($avg)));
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
#
|
||
|
|
# Return the cached data, or undef if not present...
|
||
|
|
#
|
||
|
|
# FIXME: To use multiple DBs, we'll need to cache by dbhname here...
|
||
|
|
sub QueryResultIsCached {
|
||
|
|
my ($sql, $key, $opts) = @_;
|
||
|
|
|
||
|
|
if ( $opts->{deptables} && ! $dm::use_dependent_table_strategy ) {
|
||
|
|
$dm::query_cache_misses++;
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
#
|
||
|
|
# cache entry non-existent?
|
||
|
|
#
|
||
|
|
# logger::loginfo("Checking for cache entry $key existence ...");
|
||
|
|
if ( ! exists($dm::query_cache{$key}{'data'}) ) {
|
||
|
|
$dm::query_cache_misses++;
|
||
|
|
return (undef,undef);
|
||
|
|
}
|
||
|
|
|
||
|
|
#
|
||
|
|
# cache entry expired?
|
||
|
|
#
|
||
|
|
if ( exists($dm::query_cache{$key}{'expires'}) ) {
|
||
|
|
# logger::loginfo("OK. Checking for cache entry $key expiration ...");
|
||
|
|
if ( $dm::query_cache{$key}{'expires'} <= toolbox::NowAsScalar() ) {
|
||
|
|
|
||
|
|
# logger::loginfo("Cache entry $key expired...");
|
||
|
|
$dm::query_cache_misses++;
|
||
|
|
DeleteFromCache($key);
|
||
|
|
return (undef,undef);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
#
|
||
|
|
# Dependent tables changed?
|
||
|
|
#
|
||
|
|
if ( $dm::use_dependent_table_strategy &&
|
||
|
|
exists($dm::query_cache{$key}{deptables}) ) {
|
||
|
|
|
||
|
|
# logger::loginfo("OK. Checking for stale dependent table data ...");
|
||
|
|
foreach my $deptable ( @{$dm::query_cache{$key}{deptables}} ) {
|
||
|
|
# logger::loginfo("OK. Checking staleness of $deptable ...");
|
||
|
|
if ( StaleTableData($deptable, $dm::query_cache{$key}{time}) ) {
|
||
|
|
logger::loginfo("Dependent table $deptable is stale...") if ( $dm::log_cache_actions );
|
||
|
|
$dm::query_cache_misses++;
|
||
|
|
DeleteFromCache($key);
|
||
|
|
return (undef,undef);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
#
|
||
|
|
# cache entry transaction completed/gone?
|
||
|
|
#
|
||
|
|
if ( exists($dm::query_cache{$key}{'cutoff_txid'}) && $opts->{cutoff_txid}){
|
||
|
|
logger::loginfo("OK. Checking cache entry $key txid cutoff ...");
|
||
|
|
if ( $dm::query_cache{$key}{'cutoff_txid'} < $opts->{cutoff_txid} ) {
|
||
|
|
|
||
|
|
# logger::loginfo("Cache entry $key txid cutoff $dm::query_cache{$key}{'cutoff_txid'} passed by $opts->{cutoff_txid} ...");
|
||
|
|
$dm::query_cache_misses++;
|
||
|
|
DeleteFromCache($key);
|
||
|
|
return (undef,undef);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
# logger::loginfo("OK. Returning cache hit");
|
||
|
|
$dm::query_cache{$key}{'hits'}++;
|
||
|
|
$dm::query_cache_hits++;
|
||
|
|
$dm::query_time_saved += $dm::query_cache{$key}{'query_time'};
|
||
|
|
logger::loginfo(sprintf("CACHE SAVINGS: (%d hits X %.3f secs/miss = %.3f secs saved, OVERALL: %.1f secs", $dm::query_cache{$key}{'hits'}, $dm::query_cache{$key}{'query_time'}, $dm::query_cache{$key}{'hits'} * $dm::query_cache{$key}{'query_time'}, $dm::query_time_saved)) if ( $dm::log_cache_savings );
|
||
|
|
return (1, $dm::query_cache{$key}{'data'});
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
##############################################################################
|
||
|
|
sub StaleTableData( $ $ ) {
|
||
|
|
my ($table, $cached_time) = @_;
|
||
|
|
|
||
|
|
# No data for this table?
|
||
|
|
if ( ! exists($dm::table_status{$table}) ) {
|
||
|
|
logger::loginfo("No table status info for table [$table]...unable to cache dependencies for query cache invalidation process, so no query caching on this one...");
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
my $now = toolbox::NowAsScalar();
|
||
|
|
|
||
|
|
my $status_age = $now - $dm::table_status{$table}{last_checked};
|
||
|
|
if ( $status_age > $dm::tolerable_cache_race_window ) {
|
||
|
|
|
||
|
|
logger::loginfo(sprintf("%s table status age (%.3f secs) violates race condition tolerance (%.3f secs) ... will reload latest status for %s table", $table, $status_age, $dm::tolerable_cache_race_window, $table));
|
||
|
|
loadTableChanges();
|
||
|
|
}
|
||
|
|
|
||
|
|
# Data grown stale for this table due to other updates?
|
||
|
|
if ( $dm::table_status{$table}{last_update} >= $cached_time ) {
|
||
|
|
logger::loginfo("Cached data from $table is stale...");
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
return 0;
|
||
|
|
}
|
||
|
|
|
||
|
|
##############################################################################
|
||
|
|
sub validateCachedTable {
|
||
|
|
}
|
||
|
|
|
||
|
|
##############################################################################
|
||
|
|
sub loadTableChanges {
|
||
|
|
my ($table) = @_;
|
||
|
|
|
||
|
|
logger::loginfo("Loading table_status changes...");
|
||
|
|
my $sql = qq{
|
||
|
|
SELECT id, date_part('epoch', last_update) AS "last_update", tablename
|
||
|
|
FROM table_status
|
||
|
|
WHERE last_update >= ?
|
||
|
|
};
|
||
|
|
my $params = [ $dm::last_table_status_check_time || 1 ];
|
||
|
|
if ( $table ) {
|
||
|
|
$sql .= qq{ AND tablename = ?};
|
||
|
|
push( @$params, $table );
|
||
|
|
logger::loginfo("Reloading table status changes for $table");
|
||
|
|
}
|
||
|
|
my $now = toolbox::NowAsScalar();
|
||
|
|
my $data = dm::SelectData($sql, $params);
|
||
|
|
foreach ( @$data ) {
|
||
|
|
my $table = $_->{tablename};
|
||
|
|
$dm::table_status{$table}{last_update} = $_->{last_update};
|
||
|
|
$dm::table_status{$table}{id} = $_->{id};
|
||
|
|
$dm::table_status{$table}{last_checked} = $now;
|
||
|
|
}
|
||
|
|
logger::loginfo("Done loading table_status changes...");
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
#
|
||
|
|
# Make room for $size more bytes in the cache...
|
||
|
|
#
|
||
|
|
sub PruneCache ( $ ) {
|
||
|
|
my ($size) = @_;
|
||
|
|
|
||
|
|
logger::loginfo("DB query cache is full ($dm::query_cache_size/$dm::max_cache_size)...pruning LRU queries...");
|
||
|
|
#
|
||
|
|
# LRU cache...remove LRU items until size available...
|
||
|
|
#
|
||
|
|
my $i = 1;
|
||
|
|
my $reduction = (0.2 * $dm::max_cache_size) + $size;
|
||
|
|
my $targetsize = $dm::query_cache_size - $reduction;
|
||
|
|
|
||
|
|
my $sortf = sub {
|
||
|
|
if ( exists($dm::query_cache{$a}{'time'}) &&
|
||
|
|
exists($dm::query_cache{$b}{'time'}) ) {
|
||
|
|
|
||
|
|
$dm::query_cache{$a}{'time'} <=> $dm::query_cache{$b}{'time'}
|
||
|
|
} else {
|
||
|
|
# if ( ! exists($dm::query_cache{$a}{'time'}) ) {
|
||
|
|
# logger::logwarn("Undefined 'time' value in dm::query_cache{$a}");
|
||
|
|
# }
|
||
|
|
# if ( ! exists($dm::query_cache{$b}{'time'}) ) {
|
||
|
|
# logger::logwarn("Undefined 'time' value in dm::query_cache{$b}");
|
||
|
|
# }
|
||
|
|
return 0;
|
||
|
|
}
|
||
|
|
};
|
||
|
|
|
||
|
|
foreach my $q (sort $sortf keys %dm::query_cache) {
|
||
|
|
# logger::loginfo("Pruning ". $i++. ": ". $dm::query_cache{$q}{'time'}. ", ". $dm::query_cache{$q}{'size'}); # if ( $dm::debug );
|
||
|
|
|
||
|
|
DeleteFromCache($q);
|
||
|
|
|
||
|
|
#
|
||
|
|
# Have we made enough room yet?
|
||
|
|
#
|
||
|
|
if ( $dm::query_cache_size <= $targetsize ) {
|
||
|
|
last;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub DeleteFromCache ( $ ) {
|
||
|
|
my ($key) = @_;
|
||
|
|
|
||
|
|
$dm::query_cache_size -= ($dm::query_cache{$key}{'size'} || 0);
|
||
|
|
return delete $dm::query_cache{$key};
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
# return list of dependent tables
|
||
|
|
sub getDependentTables {
|
||
|
|
my ($sql,$opts) = @_;
|
||
|
|
|
||
|
|
my $origsql = $sql;
|
||
|
|
|
||
|
|
$sql = join(/ /, $sql);
|
||
|
|
$sql =~ s/\n/ /g;
|
||
|
|
|
||
|
|
# bail if it has nested subqueries (too hard to parse)
|
||
|
|
my $selectcount = 0;
|
||
|
|
while ( $sql =~ /SELECT/gi ) {
|
||
|
|
$selectcount++;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( $selectcount > 1 ) {
|
||
|
|
logger::loginfo("Not able to parse subqueries yet (found $selectcount here)...bailing on caching this one: [$origsql]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
# slice off the SELECT clause...
|
||
|
|
$sql =~ s/^\s*SELECT\s+\S+.*\s+FROM\s+(\S+.*$)/$1/g;
|
||
|
|
$sql =~ s/(.*)WHERE\s+.*$/$1/g;
|
||
|
|
$sql =~ s/(.*)GROUP BY\s+.*$/$1/g;
|
||
|
|
$sql =~ s/(.*)HAVING\s+.*$/$1/g;
|
||
|
|
$sql =~ s/(.*)ORDER BY\s+.*$/$1/g;
|
||
|
|
$sql =~ s/(.*)LIMIT\s+.*$/$1/g;
|
||
|
|
$sql =~ s/(.*)OFFSET\s+.*$/$1/g;
|
||
|
|
$sql =~ s/;//g;
|
||
|
|
|
||
|
|
# get rid of the aliases
|
||
|
|
while ( $sql =~ s/(\s*[A-Za-z0-9_]+)\s+[A-Za-z0-9_]+,/$1,/g ) {};
|
||
|
|
while ( $sql =~ s/(\s*[A-Za-z0-9_]+)\s+[A-Za-z0-9_]+\s*$/$1/g ) {};
|
||
|
|
|
||
|
|
# punt if we don't have something that looks like a clean list...
|
||
|
|
if ( $sql !~ /^\s*([A-Za-z0-9_]+)(\s*,\s+[A-Za-z0-9_]+)*\s*$/ ) {
|
||
|
|
logger::loginfo("Unable to parse dependent tables from query: $origsql ... parsed result looks like this: [$sql]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
my @deptables = split(/,/, $sql);
|
||
|
|
|
||
|
|
foreach ( @deptables ) {
|
||
|
|
s/^\s+//g;
|
||
|
|
s/\s+$//g;
|
||
|
|
}
|
||
|
|
|
||
|
|
return \@deptables;
|
||
|
|
}
|
||
|
|
|
||
|
|
sub CacheResults {
|
||
|
|
my ($sql, $data, $sqlkey, $opts) = @_;
|
||
|
|
|
||
|
|
if ( ! $sql || ! $sqlkey ) {
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( $opts->{deptables} && ! $dm::use_dependent_table_strategy ) {
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
my $key = getCacheKey($sqlkey);
|
||
|
|
|
||
|
|
my $t0 = toolbox::NowAsScalar();
|
||
|
|
|
||
|
|
my $i = 1;
|
||
|
|
my $size = 0;
|
||
|
|
logger::loginfo("Sizing ".scalar(@$data)." rows of data...") if ( $dm::log_cache_actions );
|
||
|
|
if ( $opts->{want_array_refs} ) {
|
||
|
|
foreach my $aref ( @$data ) {
|
||
|
|
foreach my $cref ( @$aref ) {
|
||
|
|
$size += (defined($cref) ? length($cref) : 0);
|
||
|
|
$size += length($cref);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
foreach my $href ( @$data ) {
|
||
|
|
while ( my($k,$v) = each %$href ) {
|
||
|
|
$size += (defined($v) ? length($v) : 0);
|
||
|
|
$size += length($k);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
$size += length($key);
|
||
|
|
|
||
|
|
if ( $size + $dm::query_cache_size > $dm::max_cache_size ) {
|
||
|
|
logger::loginfo("Cache too big...pruning $size bytes now ...");
|
||
|
|
PruneCache($size);
|
||
|
|
}
|
||
|
|
|
||
|
|
$dm::query_cache_size += $dm::query_cache{$key}{'size'} = $size;
|
||
|
|
$dm::query_cache{$key}{'data'} = $data;
|
||
|
|
$dm::query_cache{$key}{'query_time'} = $opts->{query_time};
|
||
|
|
$dm::query_cache{$key}{'time'} = toolbox::NowAsScalar();
|
||
|
|
|
||
|
|
if ( $dm::use_dependent_table_strategy &&
|
||
|
|
$opts->{deptables} && ref($opts->{deptables}) eq 'ARRAY' ) {
|
||
|
|
|
||
|
|
#
|
||
|
|
# Dependencies explicitly specified, so use them...
|
||
|
|
#
|
||
|
|
$dm::query_cache{$key}{deptables} = [ @{$opts->{deptables}} ];
|
||
|
|
logger::loginfo("Caching ".scalar(@{$dm::query_cache{$key}{deptables}})." deptables ... ") if ( $dm::log_cache_actions );
|
||
|
|
} elsif ( $dm::use_dependent_table_strategy &&
|
||
|
|
$dm::derive_unstated_table_dependencies ) {
|
||
|
|
#
|
||
|
|
# This is where we attempt to determine the set of tables on which
|
||
|
|
# this query depends. This works great in simple cases, but needs
|
||
|
|
# refinement!
|
||
|
|
#
|
||
|
|
$dm::query_cache{$key}{deptables} = getDependentTables($sql);
|
||
|
|
if ( ! $dm::query_cache{$key}{deptables} ) {
|
||
|
|
logger::loginfo("getDependentTables() failed for query: $sql") if ( $dm::log_cache_actions );
|
||
|
|
delete $dm::query_cache{$key};
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
logger::loginfo("Dependent tables: ",@{$dm::query_cache{$key}{deptables}}) if ( $dm::log_cache_actions );
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( $opts->{cache_lifespan} ) {
|
||
|
|
$dm::query_cache{$key}{'expires'} =
|
||
|
|
$dm::query_cache{$key}{'time'} + $opts->{'cache_lifespan'};
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( $opts->{cutoff_txid} ) {
|
||
|
|
$dm::query_cache{$key}{'cutoff_txid'} = $opts->{'cutoff_txid'};
|
||
|
|
}
|
||
|
|
|
||
|
|
$sqlkey =~ s/\n/ /g;
|
||
|
|
$sqlkey =~ s/(\S)\s+(\S)/$1 $2/g;
|
||
|
|
$sqlkey =~ s/^\s+(\S)/$1/g;
|
||
|
|
$sqlkey =~ s/(\S)\s+$/$1/g;
|
||
|
|
$sqlkey = substr($sqlkey, 0, 25);
|
||
|
|
logger::loginfo("Caching $dm::query_cache{$key}{'size'}B ".($dm::query_cache{$key}{'expires'} ? "until $dm::query_cache{$key}{'expires'}" : "")." ".($dm::query_cache{$key}{'cutoff_txid'} ? "for txid $dm::query_cache{$key}{'cutoff_txid'}" : "")." under key $key for $sqlkey (cache size: $dm::query_cache_size bytes...)\n") if ( $dm::log_cache_actions ); # if ( $dm::debug );
|
||
|
|
|
||
|
|
my $et = toolbox::NowAsScalar() - $t0;
|
||
|
|
logger::loginfo("dm::CacheResults() took ".sprintf("%.3f secs", ($et))." to cache [$sqlkey]") if ( $dm::log_cache_actions );
|
||
|
|
$dm::query_cache{$key}{'caching_time'} = $et;
|
||
|
|
$dm::query_time_saved -= $et;
|
||
|
|
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
#
|
||
|
|
# Return 1 if parameter mismatch, 0 otherwise...
|
||
|
|
#
|
||
|
|
sub ParameterMismatch( $ $ ) {
|
||
|
|
my ($sql, $params) = @_;
|
||
|
|
|
||
|
|
if ( defined($params) && ref($params) ne 'ARRAY' ) {
|
||
|
|
logger::logerr("Bind params were not passed as aref");
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
my $pcount = ($sql =~ tr/?/?/);
|
||
|
|
my $bcount = (defined($params) ? scalar(@$params) : 0);
|
||
|
|
|
||
|
|
if ( $pcount != $bcount ) {
|
||
|
|
logger::logerr("Number of parameter placeholders ($pcount) does not match the number of input bind params ($bcount) for sql: [$sql], params: [".((defined($params) && (ref($params) eq 'ARRAY')) ? join($dm::cfg{paramsep},@$params) : "")."]");
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
return 0;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
#
|
||
|
|
# Select a single scalar value from a single record in the DB. If the
|
||
|
|
# query returns anything else, it is an error and the result is undef.
|
||
|
|
#
|
||
|
|
sub SelectDataValue {
|
||
|
|
my ($sql,$params,$opts) = @_;
|
||
|
|
|
||
|
|
my $data = SelectData($sql, $params, $opts);
|
||
|
|
if ( ! defined($data) ) {
|
||
|
|
logger::logerr("Query passed to SelectDataValue() returned undef. SQL: [$sql], PARAMS: [".join($dm::cfg{paramsep},@$params)."]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( ref($data) ne 'ARRAY' ) {
|
||
|
|
logger::logerr("Query passed to SelectDataValue() returned non-ARRAY ref. SQL: [$sql], PARAMS: [".join($dm::cfg{paramsep},@$params)."]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
my $count = scalar(@$data);
|
||
|
|
if ( $count != 1 ) {
|
||
|
|
logger::logerr("Query passed to SelectDataValue() returned $count records when 1 was expected. SQL: [$sql], PARAMS: [".join($dm::cfg{paramsep},@$params)."]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
my @keys = keys %{$data->[0]};
|
||
|
|
$count = scalar(@keys);
|
||
|
|
if ( $count != 1 ) {
|
||
|
|
logger::logerr("Query passed to SelectDataValue() returned $count columns when 1 was expected. SQL: [$sql], PARAMS: [".join($dm::cfg{paramsep},@$params)."]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
return $data->[0]->{$keys[0]};
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
sub SelectData {
|
||
|
|
$dm::reconnect_tries = 0;
|
||
|
|
$dm::vacuum_tries = 0;
|
||
|
|
my $data;
|
||
|
|
eval { $data = _SelectData(@_); };
|
||
|
|
if ( $@ ) {
|
||
|
|
logger::logerr($@);
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
return $data;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub Reconnect {
|
||
|
|
my ($opts) = @_;
|
||
|
|
|
||
|
|
my $dbhname = $opts->{dbhname} || $dm::cfg{dbhname};
|
||
|
|
|
||
|
|
if ( $dm::reconnect_tries++ < $dm::reconnect_limit ) {
|
||
|
|
logger::loginfo("Trying to reconnect to DB (attempt #$dm::reconnect_tries)...");
|
||
|
|
eval { CloseDB($opts) };
|
||
|
|
$dm::dbhCache{$dbhname} = GetDBH($opts);
|
||
|
|
if ( ! $dm::dbhCache{$dbhname} ) {
|
||
|
|
logger::logerr("GetDBH() FAILED...");
|
||
|
|
} else {
|
||
|
|
logger::loginfo("Hey, looks like reconnect worked...cleaning up by force just in case...");
|
||
|
|
dm::Cleanup(undef,1);
|
||
|
|
}
|
||
|
|
return $dm::dbhCache{$dbhname};
|
||
|
|
}
|
||
|
|
logger::logerr("Giving up on attempted DB reconnection after $dm::reconnect_tries tries...");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
sub getCacheKey ( $ ) {
|
||
|
|
my ($longkey) = @_;
|
||
|
|
return $longkey;
|
||
|
|
}
|
||
|
|
|
||
|
|
sub CachingIsAppropriate( $ ) {
|
||
|
|
my ($opts) = @_;
|
||
|
|
|
||
|
|
if ( $dm::query_cache_enabled &&
|
||
|
|
(($opts && $opts->{cache}) || $dm::cache_everything) ) {
|
||
|
|
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
return 0;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub _SelectData {
|
||
|
|
my ($sql,$params,$opts) = @_;
|
||
|
|
|
||
|
|
if ( ! defined($sql) ) {
|
||
|
|
logger::logerr("Undefined SQL query string; ABORTING QUERY.");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( ref($sql) ) {
|
||
|
|
logger::logerr("Non-scalar SQL query string [$sql]; ABORTING QUERY.");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
my $t0 = toolbox::TimeNow();
|
||
|
|
SetDBGSettings($opts);
|
||
|
|
|
||
|
|
my @data = ();
|
||
|
|
|
||
|
|
# FIXME: FOR DEBUGGING ONLY, THIS WILL KILL PERFORMANCE IN PRODUCTION
|
||
|
|
if ( 0 ) {
|
||
|
|
logger::loginfo("SQL: $sql");
|
||
|
|
foreach my $p ( @$params ) {
|
||
|
|
logger::loginfo("PARAM: $p");
|
||
|
|
}
|
||
|
|
}
|
||
|
|
my $pstr = (defined($params) && ref($params) eq 'ARRAY' ? join($dm::cfg{paramsep}, @$params) : "");
|
||
|
|
|
||
|
|
if ( ParameterMismatch($sql,$params) ) {
|
||
|
|
logger::logerr("Parameter mismatch for sql: [$sql] and bind params [$pstr]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
#
|
||
|
|
# See if we have this one sitting in the cache...
|
||
|
|
#
|
||
|
|
my $key = $sql . $pstr;
|
||
|
|
my ($cache_hit,$cached_data);
|
||
|
|
|
||
|
|
#
|
||
|
|
# Maybe we could use a tied IPC::Sharable implementation for sharing a
|
||
|
|
# value that indicates when child client processes need to clear their
|
||
|
|
# caches? Needs more thought...
|
||
|
|
#
|
||
|
|
if ( CachingIsAppropriate( $opts ) ) {
|
||
|
|
|
||
|
|
$key =~ s/\n//g;
|
||
|
|
$key =~ s/\s//g;
|
||
|
|
my $cachekey = getCacheKey($key);
|
||
|
|
($cache_hit,$cached_data) =
|
||
|
|
QueryResultIsCached($sql, $cachekey, $opts);
|
||
|
|
if ( $cache_hit ) {
|
||
|
|
# logger::loginfo("DB CACHE HIT ...\n") if ( $dm::show_query_cache_hitmiss || $dm::profile );
|
||
|
|
$dm::query_profile{$sql}{'hits'}++;
|
||
|
|
return $cached_data;
|
||
|
|
} else {
|
||
|
|
# logger::loginfo("DB CACHE MISS ...\n") if ( $dm::show_query_cache_hitmiss || $dm::profile );
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
logger::loginfo("SQL: [" . $sql . "], INPUT PARAMS: [" . $pstr . "]") if ( $dm::debug );
|
||
|
|
|
||
|
|
my $dbh = GetDBH($opts);
|
||
|
|
if (! $dbh) {
|
||
|
|
logger::logerr("DBI::connect() FAILED for SQL [$sql], params [$pstr], with DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "] and params ['dbi:$$dm::cfg{dbkind}:dbname=$$dm::cfg{dbhname}', '$$dm::cfg{dbuser}', '<passwd>']\n");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( $dm::explain_dm ) {
|
||
|
|
logger::loginfo("EXPLAIN ON");
|
||
|
|
eval {
|
||
|
|
$dbh->do("EXPLAIN VERBOSE $sql");
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
my $sth = $dbh->prepare( $sql );
|
||
|
|
if ( ! $sth ) {
|
||
|
|
logger::logerr("DBI::prepare() FAILED for SQL [$sql], params [$pstr], with DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "]\n");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( defined($params) ) {
|
||
|
|
my $reftype = ref $params;
|
||
|
|
if ( $reftype eq 'ARRAY' ) {
|
||
|
|
for ( my $i = 1; $i <= scalar(@$params); $i++ ) {
|
||
|
|
my $bind_value = $params->[$i-1];
|
||
|
|
logger::loginfo("Binding param #$i [$bind_value]...") if ( $dm::debug );
|
||
|
|
if ( ! defined($sth->bind_param($i, $bind_value) ) ) {
|
||
|
|
logger::logerr("DBI::bind_param($i, $bind_value) FAILED for SQL [$sql], params [$pstr], with DBI::err [" . $sth->err . "], DBI::state [" . $sth->state . "], DBI::errstr [".$sth->errstr."]\n");
|
||
|
|
return undef;
|
||
|
|
} else {
|
||
|
|
logger::loginfo("Successful bind of param #$i...") if ( $dm::debug );
|
||
|
|
}
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
logger::logerr("BAD REF TYPE $reftype\n");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
if ( ! $sth->execute() ) {
|
||
|
|
my $err = $sth->err();
|
||
|
|
my $errstr = $sth->errstr();
|
||
|
|
my $state = $sth->state();
|
||
|
|
logger::logerr("DBI::execute() FAILED in dm::_SelectData() with DBI::err [$err], DBI::state [$state], DBI::errstr [$errstr] for SQL [$sql], params [$pstr]\n");
|
||
|
|
addDBErrStr($errstr);
|
||
|
|
|
||
|
|
if ( $err == -1 && dm::Reconnect() ) {
|
||
|
|
|
||
|
|
logger::loginfo("Reconnect successful...trying query again ...");
|
||
|
|
return dm::_SelectData($sql,$params);
|
||
|
|
|
||
|
|
#
|
||
|
|
# For this particular error, mailing list discussions
|
||
|
|
# and experience suggest that vacuuming will fix it...
|
||
|
|
#
|
||
|
|
} elsif ( $err == 7 && ($errstr =~ /$dm::PGIDX_SCAN_ERR/) ) {
|
||
|
|
|
||
|
|
logger::logerr("THAT PESKY ExecInitIndexScan BUG HAS REARED ITS HEAD AGAIN...TRYING TO VACUUM...");
|
||
|
|
|
||
|
|
if ( dm::Vacuum($dbh) ) {
|
||
|
|
logger::loginfo("Vacuuming successful...retrying query on the fly...");
|
||
|
|
return dm::_SelectData($sql,$params);
|
||
|
|
} else {
|
||
|
|
logger::logerr("No luck, vacuum failed.");
|
||
|
|
}
|
||
|
|
};
|
||
|
|
logger::loginfo("DBI::execute() AND all on-the-fly recovery attempts FAILED for SQL [$sql], params [$pstr]\n");
|
||
|
|
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( $opts->{want_array_refs} ) {
|
||
|
|
while ( my $aref = $sth->fetchrow_arrayref) {
|
||
|
|
push( @data, [ @$aref ] );
|
||
|
|
}
|
||
|
|
# push columns names on as first row...
|
||
|
|
my $hdraref;
|
||
|
|
foreach my $name ( @{$sth->{NAME}} ) {
|
||
|
|
push( @$hdraref, $name );
|
||
|
|
}
|
||
|
|
unshift( @data, [ @$hdraref ] );
|
||
|
|
} else {
|
||
|
|
while ( my $href = $sth->fetchrow_hashref) {
|
||
|
|
push( @data, { %$href } );
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
my $et = toolbox::ElapsedTime($t0);
|
||
|
|
|
||
|
|
if ( ! $sth->finish() ) {
|
||
|
|
@data = ();
|
||
|
|
logger::logerr("DBI::finish() FAILED for SQL [$sql], params [$pstr], with DBI::err [" . $sth->err . "], DBI::state [" . $sth->state . "], DBI::errstr [" . $sth->errstr . "]\n");
|
||
|
|
setDBErrStr($sth->errstr);
|
||
|
|
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( $dm::debug == 1 && $dm::showdata ) {
|
||
|
|
printf STDERR "#############################################\n";
|
||
|
|
printf STDERR "RESULTS RETURNED BY SQL: [" . $sql . "], INPUT PARAMS: [" . $pstr . "]:\n";
|
||
|
|
my $i = 0;
|
||
|
|
if ( $opts->{want_array_refs} ) {
|
||
|
|
foreach my $aref ( @data ) {
|
||
|
|
printf STDERR "Row[".$i++."]";
|
||
|
|
foreach my $cref ( @$aref ) {
|
||
|
|
printf STDERR ", $cref";
|
||
|
|
}
|
||
|
|
printf STDERR "\n";
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
foreach my $href ( @data ) {
|
||
|
|
printf STDERR "Row[".$i++."]";
|
||
|
|
while ( my($k,$v) = each %$href ) {
|
||
|
|
my $key = (defined($k) ? $k : "undef");
|
||
|
|
my $val = (defined($v) ? $v : "undef");
|
||
|
|
printf STDERR ", '$key' = [$val]";
|
||
|
|
}
|
||
|
|
printf STDERR "\n";
|
||
|
|
}
|
||
|
|
}
|
||
|
|
printf STDERR "######## END OF RESULTS ####################\n";
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( CachingIsAppropriate( $opts ) &&
|
||
|
|
( ! $cache_hit || $opts->{'refresh_cache'} ) ) {
|
||
|
|
|
||
|
|
$opts->{query_time} = $et;
|
||
|
|
|
||
|
|
if ( ! CacheResults ( $sql, \@data, $key, $opts ) ) {
|
||
|
|
logger::loginfo("Unable to cache query results for $sql") if ( $dm::log_cache_actions );
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( $dm::profile ) {
|
||
|
|
print STDERR sprintf("DBI query: %.3f secs for SQL [%s]\n", $et, $sql);
|
||
|
|
}
|
||
|
|
IncQueryCount($sql,$et);
|
||
|
|
|
||
|
|
return \@data;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub InsertData { return NonSelectQuery(@_); }
|
||
|
|
sub ModifyData { return NonSelectQuery(@_); }
|
||
|
|
sub DeleteData { return NonSelectQuery(@_); }
|
||
|
|
|
||
|
|
sub NonSelectQuery {
|
||
|
|
if ( systemDown() ) {
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
$dm::reconnect_tries = 0;
|
||
|
|
my $data;
|
||
|
|
eval { $data = _NonSelectQuery(@_); };
|
||
|
|
if ( $@ ) {
|
||
|
|
logger::logerr($@);
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
return $data;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub _NonSelectQuery {
|
||
|
|
my $t0 = toolbox::TimeNow();
|
||
|
|
|
||
|
|
my ($sql,$params,$opts) = @_;
|
||
|
|
my $data = ();
|
||
|
|
|
||
|
|
SetDBGSettings($opts);
|
||
|
|
|
||
|
|
if ( ! defined($sql) || ref($sql) ) {
|
||
|
|
logger::logerr("Undefined or non-scalar SQL string passed in");
|
||
|
|
Carp::cluck("Undefined or non-scalar SQL string passed in\n");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( defined($params) && (ref($params) ne 'ARRAY') ) {
|
||
|
|
logger::logerr("Non-ARRAY ref passed as parameter list for SQL [$sql]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( ParameterMismatch($sql,$params) ) {
|
||
|
|
logger::logerr("Parameter mismatch for sql: [$sql] and bind params [".join($dm::cfg{paramsep},@$params)."]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
logger::logquery($sql,$params);
|
||
|
|
|
||
|
|
# my $pstr = (defined($params) ? join($dm::cfg{paramsep}, @$params) : '');
|
||
|
|
my $pstr = (defined($params) ? join(' | ', @$params) : '');
|
||
|
|
logger::loginfo("SQL: [" . $sql . "], INPUT PARAMS: [" . $pstr . "]") if ( $dm::debug );
|
||
|
|
|
||
|
|
my $dbh = GetDBH($opts);
|
||
|
|
if (! $dbh) {
|
||
|
|
logger::logerr("dm::GetDBH() FAILED for SQL [$sql], params [$pstr], with DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "] and params ['dbi:$dm::cfg{dbkind}:dbname=$dm::cfg{dbname}', '$dm::cfg{dbuser}', '<passwd>']\n");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( $opts->{'AUTOCOMMIT'} ) {
|
||
|
|
#
|
||
|
|
# Caller wants to reset AUTOCOMMIT in driver...
|
||
|
|
#
|
||
|
|
if ( $opts->{'AUTOCOMMIT'} == 0 ||
|
||
|
|
$opts->{'AUTOCOMMIT'} == 1 ) {
|
||
|
|
$dbh->{'AutoCommit'} = $opts->{'AUTOCOMMIT'};
|
||
|
|
} else {
|
||
|
|
logger::logerr("Illegal 'AUTOCOMMIT' value: [".
|
||
|
|
$opts->{'AUTOCOMMIT'} .
|
||
|
|
"] ... aborting query [$sql], params [$pstr]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
my $sth = $dbh->prepare( $sql );
|
||
|
|
if ( ! $sth ) {
|
||
|
|
logger::logerr("DBI::prepare() FAILED for SQL [$sql], params [$pstr], with DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "]\n");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( defined($params) ) {
|
||
|
|
my $reftype = ref $params;
|
||
|
|
if ( $reftype eq 'ARRAY' ) {
|
||
|
|
for ( my $i = 1; $i <= scalar(@$params); $i++ ) {
|
||
|
|
logger::loginfo("Binding param #$i...") if ( $dm::debug );
|
||
|
|
my $bind_value = $params->[$i-1];
|
||
|
|
if ( ! defined($sth->bind_param($i, $bind_value) ) ) {
|
||
|
|
logger::logerr("DBI::bind_param($i, $bind_value) FAILED for SQL [$sql], params [$pstr], with DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [".$DBI::errstr."]\n");
|
||
|
|
return undef;
|
||
|
|
} else {
|
||
|
|
logger::loginfo("Successful bind of param #$i...") if ( $dm::debug );
|
||
|
|
}
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
logger::logerr("BAD REF TYPE $reftype\n");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
my $rc;
|
||
|
|
|
||
|
|
$dm::pending_queries++;
|
||
|
|
|
||
|
|
if ( ! defined($rc = $sth->execute()) ) {
|
||
|
|
logger::logerr("DBI::execute() FAILED for SQL [$sql], params [$pstr], with DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "]\n");
|
||
|
|
addDBErrStr($DBI::errstr);
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
if ( $opts->{'SERIAL'} && $dm::dbtype == dm::PGSQL ) {
|
||
|
|
#
|
||
|
|
# Retrieve postgresql serial ID in the customary PG manner...
|
||
|
|
#
|
||
|
|
if ( $opts->{'TABLE'} ) {
|
||
|
|
my $ser_sql = qq{
|
||
|
|
SELECT id FROM $opts->{'TABLE'} WHERE oid = ?;
|
||
|
|
};
|
||
|
|
$rc = dm::SelectDataValue($ser_sql, [ $sth->{pg_oid_status} ]);
|
||
|
|
} else {
|
||
|
|
$rc = undef;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
if ( ! $sth->finish() ) {
|
||
|
|
$data = ();
|
||
|
|
logger::logerr("DBI::finish() FAILED for SQL [$sql], params [$pstr], with DBI::err [" . $sth->err . "], DBI::state [" . $sth->state . "], DBI::errstr [" . $DBI::errstr . "]\n");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
my $et = toolbox::ElapsedTime($t0);
|
||
|
|
if ( $dm::profile ) {
|
||
|
|
print STDERR sprintf("DBI query: %.3f secs for SQL [%s]\n", $et, $sql);
|
||
|
|
}
|
||
|
|
IncQueryCount($sql,$et);
|
||
|
|
|
||
|
|
my $retval = undef;
|
||
|
|
$retval = (defined($rc) && ($rc == '0E0') ? 0 : $rc);
|
||
|
|
|
||
|
|
if ( ! ref($retval) ) {
|
||
|
|
if ( $opts->{'SERIAL'} && $dm::dbtype == dm::PGSQL ) {
|
||
|
|
logger::loginfo("DB returned serial ID [$retval]") if $dm::debug;
|
||
|
|
} else {
|
||
|
|
logger::loginfo("DB effected $retval record(s)");
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
return ($retval);
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub SetOpts {
|
||
|
|
my ($opts) = @_;
|
||
|
|
|
||
|
|
my $dbh = GetDBH($opts);
|
||
|
|
if (! $dbh) {
|
||
|
|
logger::logerr("dm::GetDBH() FAILED to get db handle ... are we connected? DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( exists($opts->{'AUTOCOMMIT'}) ) {
|
||
|
|
#
|
||
|
|
# Caller wants to reset AUTOCOMMIT in driver...
|
||
|
|
#
|
||
|
|
if ( $opts->{'AUTOCOMMIT'} == 0 ||
|
||
|
|
$opts->{'AUTOCOMMIT'} == 1 ) {
|
||
|
|
$dbh->{'AutoCommit'} = $opts->{'AUTOCOMMIT'};
|
||
|
|
logger::loginfo("Setting AutoCommit to ".
|
||
|
|
$opts->{'AUTOCOMMIT'} ) if ( $dm::debug );
|
||
|
|
} else {
|
||
|
|
logger::logerr("Illegal 'AUTOCOMMIT' value: [".
|
||
|
|
$opts->{'AUTOCOMMIT'} .
|
||
|
|
"] ... attempt to set options failed");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub Cleanup {
|
||
|
|
my ($txid,$force) = @_;
|
||
|
|
|
||
|
|
return; # vacuuming is not safe while inserts/deletes are occurring...
|
||
|
|
|
||
|
|
#
|
||
|
|
# Only cleanup after $dm::cleanup_interval transactions...
|
||
|
|
#
|
||
|
|
# Tx IDs are sync'ed across all child apache processes...
|
||
|
|
#
|
||
|
|
print STDERR "txid = $txid, cleanup_interval = $dm::cleanup_interval\n";
|
||
|
|
if ( ! $force && (! defined($txid) || ($txid % $dm::cleanup_interval != 0)) ) {
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
|
||
|
|
my $data;
|
||
|
|
#
|
||
|
|
# Obvious race condition here, but unlikely we'll hit it as it is
|
||
|
|
# a small race window...
|
||
|
|
#
|
||
|
|
eval { $data = _Cleanup(@_); };
|
||
|
|
if ( $@ ) {
|
||
|
|
logger::logerr($@);
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
return $data;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub Vacuum ($) {
|
||
|
|
if ( systemDown() ) {
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
my ($dbh) = shift;
|
||
|
|
|
||
|
|
if ( $dm::vacuum_tries++ >= $dm::vacuum_limit ) {
|
||
|
|
logger::logwarn("Giving up on attempted DB vacuum after $dm::vacuum_tries tries...");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
my $rv;
|
||
|
|
logger::loginfo("Attempting VACUUM...");
|
||
|
|
eval { $rv = $dbh->do("VACUUM ANALYZE") };
|
||
|
|
if ( $@ ) {
|
||
|
|
logger::logerr($@);
|
||
|
|
logger::logerr("DBI::do('VACUUM') FAILED (eval block failed) with DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "]");
|
||
|
|
return undef;
|
||
|
|
} elsif ( ! defined($rv) ) {
|
||
|
|
logger::logerr("DBI::do('VACUUM') FAILED (DBI returned undef) with DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "]");
|
||
|
|
return undef;
|
||
|
|
} else {
|
||
|
|
logger::loginfo("Vacuum appears possibly successful with DBI::do() return value of [$rv] ...");
|
||
|
|
}
|
||
|
|
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub _Cleanup {
|
||
|
|
my ($opts) = @_;
|
||
|
|
|
||
|
|
my $dbh = GetDBH($opts);
|
||
|
|
if (! $dbh) {
|
||
|
|
logger::logerr("FAILED to get db handle ... are we connected? DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( ! dm::Vacuum($dbh) ) {
|
||
|
|
logger::loginfo("Attempt to vacuum database was unsuccessful (that's ok if db was locked for vacuuming by another process)...");
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
sub Commit {
|
||
|
|
my ($opts) = @_;
|
||
|
|
|
||
|
|
my $force = $opts->{force};
|
||
|
|
|
||
|
|
if ( systemDown() ) {
|
||
|
|
logger::loginfo("DB system down (".getDownFileName()." exists) ... aborting commit");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
if ( ! $dm::pending_queries && ! $force ) {
|
||
|
|
# logger::loginfo("No pending queries; commit ignored");
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
my $data;
|
||
|
|
eval { $data = _Commit($opts); };
|
||
|
|
if ( $@ ) {
|
||
|
|
logger::logerr($@);
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
return $data;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub _Commit {
|
||
|
|
my ($opts) = @_;
|
||
|
|
|
||
|
|
my $dbhname = $opts->{dbhname} || $dm::cfg{dbhname};
|
||
|
|
my $dbh = GetDBH($opts);
|
||
|
|
|
||
|
|
if (! $dbh) {
|
||
|
|
logger::logerr("FAILED to get db handle for $dbhname ... are we connected? DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
logger::logquery("COMMIT -- ($dm::pending_queries pending queries for $dbhname)");
|
||
|
|
|
||
|
|
my $rc = $dbh->commit();
|
||
|
|
|
||
|
|
if ( ! $rc ) {
|
||
|
|
logger::logerr($dbh->errstr);
|
||
|
|
} else {
|
||
|
|
$dm::pending_queries = 0;
|
||
|
|
}
|
||
|
|
|
||
|
|
return $rc;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
sub Rollback {
|
||
|
|
my ($opts) = @_;
|
||
|
|
my $force = $opts->{force};
|
||
|
|
if ( systemDown() ) {
|
||
|
|
logger::loginfo("DB system down (".getDownFileName()." exists) ... aborting rollback");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
#
|
||
|
|
# Always rollback, because some failed queries still need it...
|
||
|
|
#
|
||
|
|
if ( (! $force) && (! $dm::pending_queries) ) {
|
||
|
|
logger::loginfo("No pending queries; rollback ignored");
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
my $data;
|
||
|
|
eval { $data = _Rollback($opts); };
|
||
|
|
if ( $@ ) {
|
||
|
|
logger::logerr($@);
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
return $data;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
sub _Rollback {
|
||
|
|
my ($opts) = @_;
|
||
|
|
|
||
|
|
my $dbhname = $opts->{dbhname} || $dm::cfg{dbhname};
|
||
|
|
my $dbh = GetDBH($opts);
|
||
|
|
if (! $dbh) {
|
||
|
|
logger::logerr("FAILED to get db handle for $dbhname ... are we connected? DBI::err [". $DBI::err . "], DBI::state [". $DBI::state . "], DBI::errstr [" . $DBI::errstr . "]");
|
||
|
|
return undef;
|
||
|
|
}
|
||
|
|
|
||
|
|
logger::loginfo("Rolling back $dm::pending_queries pending queries for $dbhname");
|
||
|
|
logger::logquery("ROLLBACK -- $dbhname");
|
||
|
|
|
||
|
|
my $rc = $dbh->rollback();
|
||
|
|
|
||
|
|
if ( ! $rc ) {
|
||
|
|
logger::logerr($dbh->errstr);
|
||
|
|
} else {
|
||
|
|
$dm::pending_queries = 0;
|
||
|
|
}
|
||
|
|
|
||
|
|
return $rc;
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
END {
|
||
|
|
for my $dbhname ( keys %dm::dbhCache ) {
|
||
|
|
&dm::CloseDB( { dbhname => $dbhname } );
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
1;
|