File Coverage

blib/lib/DBIx/DBStag.pm

Criterion	Covered	Total	%
statement	10	12	83.3
branch			n/a
condition			n/a
subroutine	4	4	100.0
pod			n/a
total	14	16	87.5

# $Id: DBStag.pm,v 1.59 2008/02/06 00:50:55 cmungall Exp $
# -------------------------------------------------------
#
# Copyright (C) 2002 Chris Mungall
#
# This module is free software.
# You may distribute this module under the same terms as perl itself

#---
# POD docs at end of file
#---

package DBIx::DBStag;

use strict;
use vars qw($VERSION @ISA @EXPORT_OK %EXPORT_TAGS $DEBUG $AUTOLOAD);
use Carp;
use DBI;
use Data::Stag qw(:all);
use DBIx::DBSchema;
use Text::Balanced qw(extract_bracketed);
#use SQL::Statement;
use Parse::RecDescent;
$VERSION='0.12';


our $DEBUG;
our $TRACE = $ENV{DBSTAG_TRACE};

sub DEBUG {
    $DEBUG = shift if @_;
    return $DEBUG;
}

sub trace {
    my ($priority, @msg) = @_;
    return unless $ENV{DBSTAG_TRACE};
    print STDERR "@msg\n";
}

sub dmp {
    use Data::Dumper;
    print Dumper shift;
}

sub force {
    my $self = shift;
    $self->{_force} = shift if @_;
    return $self->{_force};
}


sub new {
    my $proto = shift;
    my $class = ref($proto) || $proto;
    my ($dbh) =
      rearrange([qw(dbh)], @_);

    my $self = {};
    bless $self, $class;
    if ($dbh) {
	$self->dbh($dbh);
    }
    $self;
}


sub connect {
    my $class = shift;
    my $dbi = shift;
    my $self;
    if (ref($class)) {
        $self = $class;
    }
    else {
        $self = {};
        bless $self, $class;
    }
    $dbi = $self->resolve_dbi($dbi);
    eval {
	$self->dbh(DBI->connect($dbi, @_));
    };
    if ($@ || !$self->dbh) {
	my $mapf = $ENV{DBSTAG_DBIMAP_FILE};
        if ($dbi =~ /^dbi:(\w+)/) {
            print STDERR <

Could not connect to database: "$dbi"

EITHER   The required DBD driver "$1" is not installed
    OR   There is no such database as "$dbi"

EOM
              ;
        }
        else {
            print STDERR <

Could not connect to database: "$dbi"

To connect to a database, you need to set the environment variable
DBSTAG_DBIMAP_FILE to the location of your DBI Stag resources file, OR
you need to specify the full dbi string of the database

A dbi string looks like this:

 dbi:Pg:dbname=foo;host=mypgserver.foo.com

A resources file provides mappings from logical names like "foo" to
full DBI locators suchas the one above

Please type "man DBI" for more information on DBI strings

If you are specifying a valid DBI locator or valid logical name and
still connect, check the database server is responding

EOM
              ;
        }
        die;
    }
    # HACK
    $self->dbh->{RaiseError} = 1;
    $self->dbh->{ShowErrorStatement} = 1;
    if ($dbi =~ /dbi:([\w\d]+)/) {
        $self->{_driver} = $1;
    }
    $self->setup;
    return $self;
}

sub resolve_dbi {
    my $self = shift;
    my $dbi = shift;
    if (!$dbi) {
	$self->throw("database name not provided!");
    }
    if ($dbi !~ /^dbi:/) {
	my $rh = $self->resources_hash;
	my $res =
	  $rh->{$dbi};
	if (!$res) {
	    if ($dbi =~ /:/) {
		$res =
		  {loc=>"$dbi"}
	    }
	    else {
		$res =
		  {loc=>"Pg:$dbi"};
	    }
	}
	if ($res) {
	    my $loc = $res->{loc};
	    if ($loc =~ /(\S+?):(\S+)\@(\S+)/) {
		my $dbms = $1;
		my $dbn = $2;
		my $host = $3;
                my $extra = '';
                if ($host =~ /(\S+?):(.*)/) {
                    $host = $1;
                    $extra = ":$2";
                }
		if ($dbms =~ /pg/i) {
		    $dbi = "dbi:Pg:dbname=$dbn;host=$host$extra";
		}
		elsif ($dbms =~ /db2/i) {
		    $dbi = "dbi:Pg:$dbn;host=$host$extra";
		}
                else {
                    # default - tested on MySQL
                    $dbi = "dbi:$dbms:database=$dbn:host=$host$extra";
                }
	    }
	    elsif ($loc =~ /(\S+):(\S+)$/) {
		my $dbms = $1;
		my $dbn = $2;
		$dbi = "dbi:$dbms:database=$dbn";
		if ($dbms =~ /pg/i) {
		    $dbi = "dbi:Pg:dbname=$dbn";
		}
	    }
	    else {
		$self->throw("$dbi -> $loc does not conform to standard.\n".
			     ":\@");
	    }
	}
	else {
	    $self->throw("$dbi is not a valid DBI locator.\n");
	}
    }
    return $dbi;
}

sub resources_hash {
    my $self = shift;
    my $mapf = $ENV{DBSTAG_DBIMAP_FILE};
    my $rh;
    if ($mapf) {
	if (-f $mapf) {
	    $rh = {};
	    open(F, $mapf) || $self->throw("Cannot open $mapf");
	    while () {
		chomp;
		next if /^\#/;
		s/^\!//;
		my @parts =split(' ', $_);
		next unless (@parts >= 3);
		my ($name, $type, $loc, $tagstr) =@parts;
		my %tagh = ();
		if ($tagstr) {
		    my @parts = split(/;\s*/, $tagstr);
		    foreach (@parts) {
			my ($t, $v) = split(/\s*=\s*/, $_);
			$tagh{$t} = $v;
		    }
		}
		$rh->{$name} =
		  {
		   %tagh,
		   name=>$name,
		   type=>$type,
		   loc=>$loc,
		   tagstr=>$tagstr,
		  };
	    }
	    close(F) || $self->throw("Cannot close $mapf");
	} else {
	    $self->throw("$mapf does not exist");
	}
    }
    return $rh;
}


sub resources_list {
    my $self = shift;
    my $rh =
      $self->resources_hash;
    my $rl;
    if ($rh) {
	$rl =
	  [map {$_} values %$rh];
    }
    return $rl;
}

sub find_template {
    my $self = shift;
    my $tname = shift;
    my $path = $ENV{DBSTAG_TEMPLATE_DIRS} || '';
    my $tl = $self->template_list;
    my ($template, @rest) = grep {$tname eq $_->name} @$tl;

    if (!$template) {
	print STDERR "\n\nI could not find the Stag SQL template called \"$tname\".\n";
	if (!$path) {
	    print STDERR <

In order to do use this or any other template, you need to set the environment
variable DBSTAG_TEMPLATE_DIRS to the directory or a set of directories
containing SQL templates. For example

  setenv DBSTAG_TEMPLATE_DIRS=".:\$HOME/my-sql-templates:/usr/share/system-sql-templates"

EOM1
;
	}
	else {
	    print STDERR <

I am looking in the following directories:

  $path

Check the contents of the directory to see if the stag sql template
you require is there, and is readable by you. Stag SQL templates
should end with the suffix ".stg"

If you wish to search other directories, set the environment variable
DBSTAG_TEMPLATE_DIRS, like this:

  setenv DBSTAG_TEMPLATE_DIRS=".:\$HOME/my-sql-templates:$path"

EOM2
;
	}
	$self->throw("Could not find template \"$tname\" in: $path");
    }
    return $template;
}

sub find_templates_by_schema {
    my $self = shift;
    my $schema = shift;
    my $tl = $self->template_list;

    my @templates = grep {$_->stag_props->tmatch('schema', $schema)} @$tl;

    return \@templates;
}

sub find_templates_by_dbname {
    my $self = shift;
    my $dbname = shift;
    my $res = $self->resources_hash->{$dbname};
    my $templates;
    if ($res) {
	my $schema = $res->{schema} || '';
	if ($schema) {
	    $templates = $self->find_templates_by_schema($schema);
	}
	else {
	    # unknown schema - show all templates
#	    $templates = $self->template_list;
	}
    }
    else {
	$self->throw("unknown db: $dbname");
    }
    return $templates;
}

sub template_list {
    my $self = shift;
    my %already_got = ();
    if (!$self->{_template_list}) {
        my $path = $ENV{DBSTAG_TEMPLATE_DIRS} || '.';
        my @dirs = split(/:/, $path);
        my @templates = ();

        foreach my $dir (@dirs) {
            foreach my $fn (glob("$dir/*.stg")) {
                if (-f $fn) {
                    require "DBIx/DBStag/SQLTemplate.pm";
                    my $template = DBIx::DBStag::SQLTemplate->new;
                    $template->parse($fn);
                    push(@templates, $template) unless $already_got{$template->name};
		    $already_got{$template->name} = 1;
                }
            }
        }
        $self->{_template_list} = \@templates;
    }
    return $self->{_template_list};
}

sub find_schema {
    my $self = shift;
    my $dbname = shift;
    my $rl = $self->resouces_list || [];
    my ($r) = grep {$_->{name} eq $_ ||
		      $_->{loc} eq $_} @$rl;
    if ($r) {
	return $r->{schema};
    }
    return;
}

sub setup {
    my $self = shift;
    return;
}

# counter
sub next_id {
    my $self = shift;
    $self->{_next_id} = shift if @_;
    $self->{_next_id} = 0 unless $self->{_next_id};
    return ++$self->{_next_id};
}


sub dbh {
    my $self = shift;
    $self->{_dbh} = shift if @_;
    return $self->{_dbh};
}

sub dbschema {
    my $self = shift;
    $self->{_dbschema} = shift if @_;
    if (!$self->{_dbschema}) {
        if (!$self->dbh) {
            confess("you must establish connection using connect() first");
        }
        $self->dbschema(DBIx::DBSchema->new_native($self->dbh));
#	my $sth = $self->dbh->table_info(undef, undef, undef, 'VIEW') or die $self->dbh->errstr;
#	use Data::Dumper;
#	print Dumper $sth->fetchall_arrayref([2,3]);
    }
    return $self->{_dbschema};
}

sub parser {
    my $self = shift;
    $self->{_parser} = shift if @_;
    if (!$self->{_parser}) {
	$self->{_parser} = Parse::RecDescent->new($self->selectgrammar());
    }
    return $self->{_parser};
}

sub warn {
    my $self = shift;
    my $fmt = shift;

    print STDERR "\nWARNING:\n";
    printf STDERR $fmt, @_;
    print STDERR "\n";
}

sub throw {
    my $self = shift;
    my $fmt = shift;

    print STDERR "\nERROR:\n";
    printf STDERR $fmt, @_;
    print STDERR "\n";
    confess;
}

sub get_pk_col {
    my $self = shift;
    my $table = shift;

    my $tableobj = $self->dbschema->table(lc($table));
    if (!$tableobj) {
        confess("Can't get table $table from db.\n".
                "Maybe DBIx::DBSchema does not work with your database?");
    }
    return $tableobj->primary_key;
}

sub is_table {
    my $self = shift;
    my $tbl = shift;
    return 1 if $self->dbschema->table($tbl);
}

sub is_col {
    my $self = shift;
    my $col = shift;
    if ($self->{_is_col_h}) {
        return $self->{_is_col_h}->{$col}
    }
    my @tablenames = $self->dbschema->tables;
    my @allcols =
      map {
          $self->get_all_cols($_);
      } @tablenames;
    my %h = map {$_=>1} @allcols;
    $self->{_is_col_h} = \%h;
    return $self->{_is_col_h}->{$col};
}

# ASSUMPTION: pk names same as fk names
sub is_fk_col {
    my $self = shift;
    my $col = shift;
    # HACK!!!
    # currently dbschema does not know about FKs
    return 1 if $col =~ /_id$/;
    if ($self->{_is_fk_col_h}) {
        return $self->{_is_fk_col_h}->{$col}
    }
    my @tablenames = $self->dbschema->tables;
    my %h = ();
    foreach (@tablenames) {
        my $pk = $self->dbschema->table($_)->primary_key;
        $h{$pk} =1 if $pk;
    }
    $self->{_is_fk_col_h} = \%h;
    return $self->{_is_fk_col_h}->{$col};
}

sub is_pk_col {
    my $self = shift;
    my $col = shift;
    if ($self->{_is_pk_col_h}) {
        return $self->{_is_pk_col_h}->{$col}
    }
    my @tablenames = $self->dbschema->tables;
    my %h = ();
    foreach (@tablenames) {
        my $pk = $self->dbschema->table($_)->primary_key;
        $h{$pk} =1 if $pk;
    }
    $self->{_is_pk_col_h} = \%h;
    return $self->{_is_pk_col_h}->{$col};
}

sub get_all_cols {
    my $self = shift;
    my $table = shift;

    my $tableobj = $self->dbschema->table(lc($table));
    if (!$tableobj) {
        confess("Can't get table $table from db.\n".
                "Maybe DBIx::DBSchema does not work with your database?");
    }
    return $tableobj->columns;
}

sub get_unique_sets {
    my $self = shift;
    my $table = shift;

    my $tableobj = $self->dbschema->table(lc($table));
    if (!$tableobj) {
        confess("Can't get table $table from db.\n".
                "Maybe DBIx::DBSchema does not work with your database?");
    }
    if ($ENV{OLD_DBIX_DBSCHEMA}) {
        return @{$tableobj->unique->lol_ref || []};
    }
    else {
        my %indices = $tableobj->indices;
        my @unique_indices = grep {$_->unique} values %indices;
        return map {$_->columns} @unique_indices;
    }
}

sub mapconf {
    my $self = shift;
    my $fn = shift;
    my $fh = FileHandle->new($fn) || confess("cannot open $fn");
    my @mappings = <$fh>;
    $fh->close;
    $self->mapping(\@mappings);
}

sub mapping {
    my $self = shift;
    if (@_) {
	my $ml = shift;
	my @nu =
	  map {
	      if (ref($_)) {
		  Data::Stag->nodify($_);
	      }
	      else {
		  if (/^(\w+)\/(\w+)\.(\w+)=(\w+)\.(\w+)/) {
		      Data::Stag->new(map=>[
					    [fktable_alias=>$1],
					    [table=>$2],
					    [col=>$3],
					    [fktable=>$4],
					    [fkcol=>$5]
					   ]);
		  }
		  elsif (/^(\w+)\.(\w+)=(\w+)\.(\w+)/) {
		      Data::Stag->new(map=>[
					    [table=>$1],
					    [col=>$2],
					    [fktable=>$3],
					    [fkcol=>$4]
					   ]);
		  }
		  elsif (/^parentfk:(\w+)\.(\w+)/) {
		      Data::Stag->new(parentfk=>[
                                                 [table=>$1],
                                                 [col=>$2],
                                                ]);
		  }
		  else {
		      confess("incorrectly specified mapping: $_".
			      "(must be alias/tbl.col=ftbl.fcol)");
		      ();
		  }
	      }
	  } @$ml;
	$self->{_mapping} = \@nu;
    }
    return $self->{_mapping};
}

sub guess_mapping {
    my $self = shift;
    my $dbschema = $self->dbschema;

    $self->mapping([]);
    my %th =
      map { $_ => $dbschema->table($_) } $dbschema->tables;
    foreach my $tn (keys %th) {
        my @cns = $th{$tn}->columns;
        foreach my $cn (@cns) {
            my $ftn = $cn;
            $ftn =~ s/_id$//;
            if ($th{$ftn}) {
                push(@{$self->mapping},
                     Data::Stag->new(map=>[
                                           [table=>$tn],
                                           [col=>$cn],
                                           [fktable=>$ftn],
                                           [fkcol=>$cn]
                                          ]));
            }
        }
    }
}

sub linking_tables {
    my $self = shift;
    $self->{_linking_tables} = {@_} if @_;
    return %{$self->{_linking_tables} || {}};
}

sub add_linking_tables {
    my $self = shift;
    my %linkh = $self->linking_tables;
    return unless %linkh;
    my $struct = shift;
    foreach my $ltname (keys %linkh) {
        my ($t1, $t2) = @{$linkh{$ltname}};
        $struct->where($t1,
                       sub {
                           my $n=shift;
                           my @v = $n->getnode($t2);
                           return unless @v;
                           $n->unset($t2);
                           my @nv =
                             map {
                                 $n->new($ltname=>[$_]);
                             } @v;
#                           $n->setnode($ltname,
#                                       $n->new($ltname=>[@v]));
                           foreach (@nv) {
                               $n->addkid($_);
                           }
                           0;
                       });
    }
    return;
}

# ----------------------------------------

sub elt_card {
    my $e = shift;
    my $c = '';
    if ($e =~ /(.*)([\+\?\*])/) {
        ($e, $c) = ($1, $2);
    }
    # make the element RDB-safe
    $e =~ s/\-//g;
    return ($e, $c);
}

sub source_transforms {
    my $self = shift;
    $self->{_source_transforms} = shift if @_;
    return $self->{_source_transforms};
}

sub autotemplate {
    my $self = shift;
    my $schema = shift;
    return () unless grep {!stag_isterminal($_)} $schema->subnodes;
    my @J = ();
    my @W = ();
    my @EXAMPLE = ();
    my ($tname) = elt_card($schema->element);
    my %joinpaths = ();

    $schema->iterate(sub {
			 my $n = shift;
			 my $parent = shift;
			 my ($tbl, $card) = elt_card($n->element);
			 if (!$parent) {
			     push(@J, $tbl);
#			     $joinpaths{$tbl} = $tbl;
			     return;
			 }
			 my ($ptbl) = elt_card($parent->element);
			 if (stag_isterminal($n)) {
			     my $v = $ptbl.'_'.$tbl;
			     my $w = "$ptbl.$tbl => \&$v\&";
			     if ($ptbl eq $tname) {
				 push(@W,
				      "[ $w ]");
			     }
			     else {
				 my $pk = $tname.'_id';
				 my $subselect =
				   "SELECT $pk FROM $joinpaths{$ptbl}".
				     " WHERE $w";
				 push(@W,
				      "[ $pk IN ($subselect) ]");
			     }
			     # produce example formula for non-ints
			     if ($n->data eq 's') {
				 push(@EXAMPLE,
				      "$v => SELECT DISTINCT $tbl FROM $ptbl");
			     }
			 }
			 else  {
			     my $jtype = 'INNER JOIN';
			     if ($card eq '*' || $card eq '?') {
				 $jtype = 'LEFT OUTER JOIN';
			     }
			     my $jcol = $ptbl.'_id';
			     push(@J,
				  "$jtype $tbl USING ($jcol)");
			     if ($joinpaths{$ptbl}) {
				 $joinpaths{$tbl} =
				   "$joinpaths{$ptbl} INNER JOIN $tbl USING ($jcol)";
			     }
			     else {
				 $joinpaths{$tbl} = $tbl;
			     }
			 }
			 return;
		     });
    my $from = join("\n  ", @J);
    my $where = join("\n  ", @W);
    my $nesting = $schema->duplicate;
    $nesting->iterate(sub {
			  my $n = shift;
			  if (stag_isterminal($n)) {
			      return;
			  }
			  my ($tbl, $card) = elt_card($n->element);
			  $n->element($tbl);
			  my @sn = $n->kids;
			  @sn =
			    grep {
				my ($tbl, $card) = elt_card($_->element);
				$_->element($tbl);
				!stag_isterminal($_)
			    } @sn;
			  if (@sn) {
			      $n->kids(@sn);
			  }
			  else {
			      $n->data([]);
			  }
		      });
    $nesting = Data::Stag->new(set=>[$nesting]);
    my $nstr = $nesting->sxpr;
    $nstr =~ s/^\'//;
    my $tt =
      join("\n",
	   ":SELECT *",
	   ":FROM $from",
	   ":WHERE $where",
	   ":USE NESTING",
	   "$nstr",
	   "",
	   "// ---- METADATA ----",
	   "schema:",
	   "desc: Fetches $tname objects",
	   "      This is an AUTOGENERATED template",
	   "",
	   (map {
	       "example_input: $_"
	   } @EXAMPLE),
	  );

#    my $template = DBIx::DBStag::SQLTemplate->new;
    my @sn = $schema->subnodes;
    my @tts = ();
    push(@tts, $self->autotemplate($_)) foreach @sn;
    return ([$tname=>$tt], @tts);
}

sub autoddl {
    my $self = shift;
    my $stag = shift;
    my $link = shift;
    $stag->makeattrsnodes;
    my $schema = $stag->autoschema;
    $self->source_transforms([]);;
    $self->_autoddl($schema, undef, $link);
}

sub _autoddl {
    my $self = shift;
    my $schema = shift;
    my $parent = shift;
    my $link = shift || [];   # link tables
    my $tbls = shift || [];
    my @sn = $schema->subnodes;
    my ($tbl, $card) = elt_card($schema->element);
    my @cols = (sprintf("%s_id serial PRIMARY KEY NOT NULL", $tbl));
    my $casc = " ON DELETE CASCADE";
    foreach (grep {stag_isterminal($_)} @sn) {
        my ($col, $card) = elt_card($_->element);
        my $pk = '';
        if ($col eq $tbl.'_id') {
            shift @cols;
            $pk = ' PRIMARY KEY';
        }
        if ($card =~ /[\+\*]/) {
	    my $new_name =  sprintf("%s_%s", $tbl, $col);
	    my $tf = ["$tbl/$col", "$new_name/$col"];
	    push(@{$self->source_transforms}, $tf);
	    $_->name($new_name);
	    $_->data([[$col => $_->data]]);
#	    $self->throw("In the source data, '$col' is a multivalued\n".
#			 "terminal (data) node. This is difficult to transform");
        }
        else {
#            my $isnull = $card eq '?' ? '' : ' NOT NULL';
            my $isnull = '';
            push(@cols,
                 sprintf("%s %s$isnull$pk",
                         $col, $_->data));
        }
    }
    if ($parent) {
        my ($pn) = elt_card($parent->element);
        push(@cols,
             sprintf("%s_id INT", $pn));
        push(@cols,
             sprintf("FOREIGN KEY (%s_id) REFERENCES $pn(%s_id)$casc", $pn, $pn));
    }

    my $mapping = $self->mapping || [];

    if (grep {$_ eq $tbl} @$tbls) {
#	$self->throw("$tbl has >1 parent - you need to\n".
#		     "transform input data");
	return "";
    }
    push(@$tbls, $tbl);

    my $post_ddl = '';
    my $pre_ddl = '';
    foreach (grep {!stag_isterminal($_)} @sn) {
        # check for cases where we want to include FK to subnode
        my ($map) =
          grep {
              $_->name eq 'map' &&
                ($_->get_table eq $tbl &&
                 ($_->get_fktable_alias eq $_->element ||
                  $_->get_fktable eq $_->element))
              } @$mapping;
	# linking tables
        if ($map ||
            grep {$_ eq $tbl} @$link) {
            my $ftbl = $_->element;
            push(@cols,
                 sprintf("%s_id INT", $ftbl));
            push(@cols,
                 sprintf("FOREIGN KEY (%s_id) REFERENCES $ftbl(%s_id)$casc", $ftbl, $ftbl));
            $pre_ddl .= $self->_autoddl($_, undef, $link, $tbls);

        }
        else {
            $post_ddl .= $self->_autoddl($_, $schema, $link, $tbls);
        }

    }
    my $ddl =
      sprintf("CREATE TABLE $tbl (\n%s\n);\n\n",
              join(",\n", map {"    $_"} @cols));

    return $pre_ddl . $ddl . $post_ddl;;
}

# ----------------------------------------
# CACHE METHODS
#
# we keep a cache of what is stored in
# each table
#
# cache->{$element}->{$key}->{$val}
# ----------------------------------------

# list of table names that should be cached
sub cached_tables {
    my $self = shift;
    $self->{_cached_tables} = shift if @_;
    return $self->{_cached_tables};
}

sub is_caching_on {
    my $self = shift;
    my $element  = shift;
    $self->{_is_caching_on} = {}
      unless $self->{_is_caching_on};
    if (@_) {
        $self->{_is_caching_on}->{$element} = shift;
    }
    return $self->{_is_caching_on}->{$element};
}

sub query_cache {
    my $self = shift;
    my $element = shift;
    my $constr = shift;
    my $update_h = shift;
    my @keycols = sort keys %$constr;
    my $cache = $self->get_tuple_idx($element, \@keycols);
    my $valstr = join("\t", map {$constr->{$_}} @keycols);
#    use Data::Dumper;
#    print Dumper $cache;
    if ($update_h) {
        my $current_h = $cache->{$valstr} || {};
        $current_h->{$_} = $update_h->{$_} foreach keys %$update_h;
        $cache->{$valstr} = $current_h;
    }
    return $cache->{$valstr};
}

sub insert_into_cache {
    my $self = shift;
    my $element = shift;
    my $insert_h = shift;
    my $usets = shift;
    foreach my $uset (@$usets) {
        my @undef = grep {!defined $insert_h->{$_}} @$uset;
        if (@undef) {
            my @defined = grep {defined $insert_h->{$_}} @$uset;
            trace(1,
                  "undefined column in unique key: @$uset IN $element/[@$uset] ".
                  join('; ',
                       map {"$_=$insert_h->{$_}"} @defined,
                      )
                 ) if $TRACE;
            # cannot cache undefined values
            next;
        }
        my $cache = $self->get_tuple_idx($element, $uset);
        my $valstr = join("\t", map {$insert_h->{$_}} sort @$uset);
        $cache->{$valstr} = $insert_h;
    }
    return 1;
}

sub update_cache {
    my $self = shift;
    my $element = shift;
    my $store_hash = shift;
    my $unique_constr = shift;

    my $tuple = $self->query_cache($element,
                                   $unique_constr,
                                   $store_hash);
    return;
}

sub get_tuple_idx {
    my $self = shift;
    my $element = shift;
    my $ukey = shift;
    my @keycols = @$ukey;
    @keycols = sort @keycols;
    @keycols || die;

    my $cache = $self->cache;
    if (!$cache->{$element}) {
        $cache->{$element} = {};
    }
    my $eltcache = $cache->{$element};
    # we just use a flat perl hash - flatten the list of unique cols
    # to a string with spaces between
    my $k = "@keycols";
    if (!$eltcache->{$k}) {
        $eltcache->{$k} = {};
    }

    return $eltcache->{$k};
}

sub cache_summary {
    my $self = shift;
    my $s = Data::Stag->new(cache_summary=>[]);
    my $cache = $self->cache || {};
    my @elts = keys %$cache;
    foreach my $elt (@elts) {
        my $cnode = $cache->{$elt} || {};
        my @keys = keys %$cnode;
        $s->add($elt=>[map {[$_=>scalar(keys %{$cnode->{$_}})]} @keys]);
    }
    return $s;
}

sub cache {
    my $self = shift;
    $self->{_cache} = shift if @_;
    $self->{_cache} = {} unless $self->{_cache};
    return $self->{_cache};
}

sub clear_cache {
    my $self = shift;
    $self->cache({});
}

# ---- END OF CACHE METHODS ----

# set this if we are loading a fresh/blank slate DB
# (will assume database is empty and not check for
#  existing tuples)
sub policy_freshbulkload {
    my $self = shift;
    $self->{_policy_freshbulkload} = shift if @_;
    return $self->{_policy_freshbulkload};
}
sub noupdate_h {
    my $self = shift;
    $self->{_noupdate_h} = shift if @_;
    return $self->{_noupdate_h} || {};
}
sub tracenode {
    my $self = shift;
    $self->{_tracenode} = shift if @_;
    return $self->{_tracenode};
}

sub mapgroups {
    my $self = shift;
    if (@_) {
        $self->{_mapgroups} = [@_];
        $self->{_colvalmap} = {}
          unless $self->{_colvalmap};
        foreach my $cols (@_) {
            my $h = {};
            foreach (@$cols) {
                $self->{_colvalmap}->{$_} = $h;
            }
        }
    }
    return @{$self->{_mapgroups} || []};
}

# DEPRECATED
sub get_mapping_for_col {
    my $self = shift;
    my $col = shift;
    $self->{_colvalmap}->{$col} = {}
      unless $self->{_colvalmap}->{$col};
    return $self->{_colvalmap}->{$col};
}

# mapping of Old ID => New ID
# IDs are assumed to be global across ALL tables
sub id_remap_idx {
    my $self = shift;
    if (@_) {
        $self->{_id_remap_idx} = shift;
    }
    else {
        $self->{_id_remap_idx} = {}
          unless $self->{_id_remap_idx};
    }
    return $self->{_id_remap_idx};
}

# do the PK values in the XML represent the actual
# internal db ids, or are they local to the document?
# if the latter then we will create a id_remap_idx
sub trust_primary_key_values {
    my $self = shift;
    $self->{_trust_primary_key_values} = shift if @_;
    return $self->{_trust_primary_key_values};
}


sub make_stag_node_dbsafe {
    my $self = shift;
    my $node = shift;
    my $parent = shift;
    my $name = $node->name;
    # CJM 2007-03-05
    #return if $name eq '@'; # leave attrs alone
    if ($name eq '@') {
        # descend into attrs
        $parent->data([grep {$_->name ne '@'} @{$parent->data},@{$node->data}]);
        return;
    }
    my $safename = $self->dbsafe($name,$parent);
    if ($name ne $safename) {
	$node->name($safename);
    }
    my @kids = $node->kids;
    foreach (@kids) {
	$self->make_stag_node_dbsafe($_,$node) if ref $_;
    }
    return;
}
sub dbsafe {
    my $self = shift;
    my $name = shift;
    my $parent = shift;
    $name = lc($name);
    # dbstag is designed for stag-like xml; no mixed attributes
    # however, we do have basic checks for mixed attributes
    if ($name eq '.') {
        $name = $parent->name.'_data'; # TODO - allow custom column
    }
    $name =~ tr/a-z0-9_//cd;
    return $name;
}

# cache the attribute nodes as they are parsed
#sub current_attribute_node {
#    my $self = shift;
#    $self->{_current_attribute_node} = shift if @_;
#    return $self->{_current_attribute_node};
#}

# lookup table; macro ID => internal database ID
sub macro_id_h {
    my $self = shift;
    $self->{_macro_id_h} = shift if @_;
    $self->{_macro_id_h} = {}
      unless $self->{_macro_id_h};
    return $self->{_macro_id_h};
}

# xort-style XML; set if an attribute is encountered
sub xort_mode {
    my $self = shift;
    $self->{_xort_mode} = shift if @_;
    return $self->{_xort_mode};
}


#'(t1
#  (foo x)
#  (t2
#   (bar y)))
#
# '(fk
#   (table t2)
#   (ftable t1))
#
# alg: store t1, then t2

#  '(t1
#    (foo x)
#    (t1_t2
#     (t2
#      (bar y))))
#
# '(fk
#   (table t1_t2)
#   (ftable t1))
# '(fk
#   (table t1_t2)
#   (ftable t2))
#
#
# alg: store t1, hold on t1_t2, store t2

#   '(t1
#     (foo x)
#     (blah
#      (t2
#       (bar y))))
#
# '(fk
#   (table t1)
#   (fktable t2)
#   (fktable_alias "blah")
#   (fk  "blah_id"))

# alg: store t2, store t1

# if set, will ensure that tbl/col names are transformed to be safe
sub force_safe_node_names {
    my $self = shift;
    $self->{_force_safe_node_names} = shift if @_;
    return $self->{_force_safe_node_names};
}


# recursively stores a Data::Stag tree node in the database
sub storenode {
    my $self = shift;
    my $node = shift;
    my @args = @_;
    my $dupnode = $node->duplicate;

    $self->make_stag_node_dbsafe($dupnode,'')
      if $self->force_safe_node_names;
    $self->add_linking_tables($dupnode);
    $self->_storenode($dupnode,@args);
}

sub _storenode {
    my $self = shift;
    my $node = shift;
    my $opt = shift;
    if (!$node) {
        confess("you need to pass in a node!");
    }
    my $element = $node->element;
    return unless $node->kids;
    if ($element eq 'dbstag_metadata') {
        my @maps = $node->get_map;
        $self->mapping(\@maps);
        my @links = $node->get_link;
        if (@links) {
            my %h =
              map {
                  ($_->sget_table => [$_->sget_from, $_->sget_to])
              } @links;
            $self->linking_tables(%h);
        }
        return;
    }

    # sql can be embedded as <_sql> tags
    if ($element eq '_sql') {
        $self->_execute_sqlnode($node);
        return;
    }

    # check for XORT-style attributes
#    if ($element eq '@') {
#        # is this check required???
#        $self->current_attribute_node($node);
#        $self->xort_mode(1);
#        return;
#    }
    my $current_attribute_node;
    unless ($node->isterminal) {
        my @kids = $node->kids;
        my $changed = 0;
        @kids =
          map {
              if ($_->element eq '@') {
                  $self->xort_mode(1);
                  $current_attribute_node = $_;
                  $changed = 1;
                  trace(0, "GOT ATTR NODE");
                  ();  # omit
              }
              else {
                  $_;    # unchanged
              }
          } @kids;
        $node->kids(@kids) if $changed;
    }

    my $operation; # directive: force/update/lookup
    if ($current_attribute_node){
        $operation =
          $current_attribute_node->sget_op;
    }

    trace(0, "STORING $element\n", $node->xml) if $TRACE;
    my $tracenode = $self->tracenode || '';
    my $tracekeyval;
    if ($tracenode && $tracenode =~ /^(\w+)\/(.*)/) {
        my $nn = $1;
        my $tag = $2;
        if ($nn eq $element) {
            $tracekeyval = $node->get($tag);
        }
    }

    my $dbh = $self->dbh;
    my $dbschema = $self->dbschema;

    my $is_caching_on = $self->is_caching_on($element) || 0;

    my $mapping = $self->mapping || [];

    # each relation has zero or one primary keys;
    # primary keys are assumed to be single-column
    my $pkcol = $self->get_pk_col($element);
    trace(0, "PKCOL: $pkcol") if $TRACE;

    # DBIx::DBSchema metadata
    my $tableobj = $dbschema->table($element);

    # -- PRE-STORE CHILD NON-TERMINALS --
    # before storing this node, we need to
    # see if we first need to store any child
    # non-terminal nodes (in order to get their
    # primary keys, to use as foreign keys in
    # the current relation)

    # store non-terminal subnodes first
    my @ntnodes = $node->ntnodes;

    # keep track of nodes that have been assigned xort-style
    my %assigned_node_h;

    # GET INFORMATION FROM SUPER-NODE
    #   some nodes may have been assigned by the calling process
    #   (eg if the supernode is refered to by a fk from the current table)
    # this hash maps element names to a boolean;
    # this is ONLY used in conjunction with xort-style xml
    #  we set this when we want to make sure that an element value is
    #  NOT macro-expanded by the expansion code
    %assigned_node_h = %{$opt->{assigned_node_h} || {}};

    # the primary key value of the supernode
    my $parent_pk_id = $opt->{parent_pk_id};
    # the element type of the supernode
    my $parent_element = $opt->{parent_element};
    # -- end of info from super-node

    # PRE-STORE
    # look through the current node's children;
    #  + some of these will be nodes that must be pre-stored BEFORE
    #    the current node (because the current node has a fk to them)
    #  + some of these will be nodes that must be post-stored AFTER
    #    the current node (because they have an fk to the current node)
    #
    #  one or other of these situations must be true - otherwise
    # nodes should not be nested!
    my @delayed_store = ();       # keep track of non-pre-stored nodes
    foreach my $nt (@ntnodes) {
        # First check for XORT-STYLE
        # xort-style XML; nodes can be nested inside a non-terminal
        # node corresponding to a FK column
        # eg
        #
        #    foo
        #
        #
        # here, what looks like a non-terminal node should actually

        # check all sub-nodes; if any of them are nonterminal and correspond
        # to a column (not a table) then add the sub-node and use the pk id
        # as the returned value
        # note: we have to explicitly check the col is not also a table
        # since some dbs (eg go db) have col names the same as tbl names
        if ($self->is_col($nt->name) &&
            !$nt->isterminal &&
            !$self->is_table($nt->name)) {
            my @kids = $nt->kids;
            if (@kids != 1) {
                $self->throw("non-terminal pk node should have one subnode only; ".
                             $nt->name." has ".scalar(@kids));
            }
            my $sn_val = $self->_storenode(shift @kids);
            if (!defined($sn_val)) {
                $self->throw("no returned value for ".$nt->name);
            }
            # TRANSFORM NODE: non-terminal to terminal
            # replace node with return pk ID value
            $nt->data($sn_val);

            # do NOT try and expand the value assigned to this
            # node with a xort-macro expansion later on
            $assigned_node_h{$nt->name} = 1;
            trace(0, "ASSIGNED NON-MACRO ID for ".$nt->name." TO $sn_val") if $TRACE;

            # skip this ntnode - it is now a tnode
            next;
        }
        # -- END OF xort-style check

        # we want to PRE-STORE any ntnodes that
        # are required for foreign key relationships
        # within this node;
        # ie this node N1 has a foreign key "fk_id" that
        # points to ntnode N2.
        # if there is an intermediate alias element in
        # between then we need to store the ntnode too
        #
        # check for either of these conditions
        my ($map) =
          grep {
              $_->get_table &&
                $_->get_table eq $element &&
                ($_->get_fktable_alias &&
                 $_->get_fktable_alias eq $nt->element ||
                 ($_->get_fktable &&
                  $_->get_fktable eq $nt->element && !$_->get_fktable_alias))
            } @$mapping;
        # check to see if sub-element has FK to this element
        if (!$map) {
#            my $subtable = $dbschema->table($nt->element);
            my $table = $dbschema->table($element);
            my $ntelement = $nt->element;
            my $subpkcol = $self->get_pk_col($ntelement);

            my @cns = $table->columns;

            my $cn;    # col name (FK in current element)
            my $fcn;   # foreign col name (PK in sub element)

            # HACK - ASSUME NATURAL JOIN
            # for example, a FK: person.dept_id => dept.dept_id
            if ($subpkcol ne 'id') {
                foreach (@cns) {
                    if ($_ eq $subpkcol) {
                        $cn = $_;
                        $fcn = $_;
                    }
                }
            }

            # second chance; allow base "id" style
            # for example, a FK: person.dept_id => dept.id
            # via ...
            if (!$cn) {
                if ($subpkcol eq 'id') {
                    foreach (@cns) {
                        if ($_ eq $ntelement."_id") {
                            $cn = $_;
                            $fcn = 'id';
                        }
                    }
                }
            }
            if ($cn) {
                $map =
                  Data::Stag->new(map=>[
                                        [table=>$element],
                                        [col=>$cn],
                                        [fktable=>$nt->element],
                                        [fkcol=>$fcn]
                                       ]);
            }
        }

        # if $map is set, then we have to pre-store this subnode
        if ($map) {
            # 1:many between this and child
            # (eg this has fk to child)
            # store child before this;
            # use fk in this
            my $fktable = $map->get_fktable;

            my $col = $map->get_col || $self->get_pk_col($fktable);

            # aliases map an extra table
            # eg table X col X.A => Y.B
            # fktable_alias = A
            my $fktable_alias = $map->get_fktable_alias;
            my $orig_nt = $nt;

            # if we have an alias, it means the actual node
            # we want to store is one beneath the alias;
            # eg ..
            # we want to actually store the node foo2
            if ($fktable_alias) {
                my @nts = $nt->sgetnode($map->sget_fktable);
                if (!@nts) {
                    print STDERR $nt->sxpr;
                    confess("could not get node for: ".$map->sget_fktable);
                }
                if (@nts > 1) {
                    print STDERR $nt->sxpr;
                    confess("multiple nodes for: ".$map->sget_fktable);
                }
                $nt = shift @nts;
                if (!$nt) {
                    print STDERR $map->sxpr;
                    print STDERR $orig_nt->sxpr;
                    confess("bad nodes for: ".$map->sget_fktable);
                }
            }
            my $fk_id = $self->_storenode($nt);
            if (!defined($fk_id)) {
                confess("ASSERTION ERROR: could not get foreign key val\n".
                        "trying to store: $element\n".
                        "no fk returned when storing: $fktable");
            }
            trace(0, "SETTING $element.$col=$fk_id [via ".$orig_nt->element."]") if $TRACE;
            $node->set($col, $fk_id);
            $node->unset($orig_nt->element);

            # do NOT try and expand the value assigned to this
            # node with a xort-macro expansion later on
            $assigned_node_h{$col} = 1;
            trace(0, "ASSIGNED NON-MACRO ID for ".$col) if $TRACE;
        }
        else {
            # 1:many between child and this
            # (eg child has fk to this)
            # store child after
            trace(0, "WILL STORE LATER:\n", $nt->xml) if $TRACE;
            $node->unset($nt->element);
            push(@delayed_store, $nt);
        }
#        $node->unset($nt->element); # clear it
    }
    # --- done storing kids

    # --- replace *IDs ---
    # dbstag XML allows placeholder values in primary key cols
    # (for now, PKs are always assumed to be autoincrement/serial ints)
    # placeholder PKs get remapped to a new autogenerated ID
    # all FKs referring to this get remapped too
    my @tnodes = $node->tnodes; # terminal nodes mapped to columns in db
    my %remap = ();   # indexed by column name; new PK value
    if (!$self->trust_primary_key_values) {
        foreach my $tnode (@tnodes) {
            # foreign keys in XORT mode - replace macro ID with
            # actual database foreign key value
            if ($self->is_fk_col($tnode->name) && $self->xort_mode) {
                my $v = $tnode->data;

                # -- CHECK FOR MACRO EXPANSION (XORT-STYLE) --
                # IF this tnode was originally an ntnode that
                # was collapsed to a pk val, xort style, do not
                # try and map it to a previously assigned macro
                # EXAMPLE:
                #   we start with A
                #   we collapse too $v
                if ($assigned_node_h{$tnode->name}) {
                    trace(0, "ALREADY CALCULATED; not a Macro ID:$v;; in $element/".$tnode->name) if $TRACE;
                    # DO NOTHING
                }
                else {  # NOT ASSIGNED
                    my $actual_id =
                      $self->macro_id_h->{$v};
                    if (!defined($actual_id)) {
                        $self->throw("XORT-style Macro ID:$v is undefined;; in $element/".$tnode->name);
                    }
                    $tnode->data($actual_id);
                }
                # -- END OF MACRO EXPANSION --
            }
            elsif ($tnode->name eq $pkcol) {
                my $v = $tnode->data;
                trace(0, "REMAP $pkcol: $v => ? [do not know new value yet]") if $TRACE;
                $remap{$tnode->name} = $v; # map after insert/update
                $node->unset($tnode->name); # discard placeholder
            } else {
                if ($self->is_fk_col($tnode->name)) {
                    # hack!! need proper FK refs...; DBSchema wont do this
                    my $colvalmap = $self->id_remap_idx;
                    #my $colvalmap = $self->get_mapping_for_col($nt->elememt);
                    if ($colvalmap) {
                        my $v = $tnode->data;
                        my $nv = $colvalmap->{$v};
                        if ($nv) {
                            trace(0, "remapping $v => $nv") if $TRACE;
                            $tnode->data($nv);
                        }
                    }
                }
            }
        }
    }  # -- end of ID remapping

    # --- Get columns that need updating/inserting ---
    # turn all remaining tag-val pairs into a hash
    my %store_hash = $node->pairs;

    # All columns to be stored should be terminal nodes
    # in the Stag tree; if not there is a problem
    my @refcols = grep { ref($store_hash{$_}) } keys %store_hash;
    if (@refcols) {
        foreach (@$mapping) {
            trace(0, $_->sxpr) if $TRACE;
        }
        confess("I can't store the current node; ".
                "These elements need to be mapped via FKs: ".
                join(', ', map {"\"@refcols\""} @refcols).
                "\n\nPerhaps you need to specify more schema metadata?");
    } # -- end of sanity check

    # each relation has zero or more unique keys;
    # unique keys may be compound (ie >1 column)
    my @usets = $self->get_unique_sets($element);
    trace(0, "USETS: ", map {"unique[ @$_ ]"} @usets) if $TRACE;

    # get all the columns/fields/attributes of this relation
    my @cols = $self->get_all_cols($element);
    trace(0, "COLS: @cols") if $TRACE;

    # store_node() will either perform an update or
    # an insert. if we are performing an update, we
    # need a query constraint to determine which row
    # to update.
    #
    # this hash is used to determine the key/val pairs
    my %unique_constr;

    # this is the value of the primary key of
    # the inserted/update row
    my $id;

    # if this relation has a primary key AND the stag node
    # being stored has the value of this column set, THEN
    # use this as the update constraint
    if (0 && $pkcol) {
        my $pk_id;
        $pk_id = $node->get($pkcol);
        if ($pk_id) {
            # unset the value of the pk in the node; there
            # is no point setting this in the UPDATE as it
            # is already part of the update constraint
            $node->unset($pkcol);

            # set the update constraint based on the PK value
            %unique_constr = ($pkcol => $pk_id);

            # return this value at the end
            $id = $pk_id;
            trace(0, "SETTING UPDATE CONSTR BASED ON PK $pkcol = $pk_id") if $TRACE;
        }
    } # -- end of xxxx


    #        foreach my $sn ($node->kids) {
    #            my $name = $sn->element;
    #            my $nu_id = $self->id_mapping($name, $sn->data);
    #            # do the old 2 nu mapping
    #            # (the ids in the xml are just temporary
    #            #  for internal consistency)
    #            $sn->data($nu_id) if $nu_id;
    #        }

    if (0) {
        # ---- EXPERIMENTAL ----
        # if no unique keys are provided, assume that all
        # non-PK columns together provide a compound unique key
        # <> expedient for now!
        if (!@usets) {
            @usets = ( [grep {$_ ne $pkcol} @cols] );
        }
    }
    if ($pkcol) {
        # make single PK the first unique key set;
        # add to beginning as this is the most efficient
        unshift(@usets, [$pkcol]);
    }

    # get the column to select to get the pk for this element
    my $select_col = $pkcol;

    # -------- find update constraint by unique keys ----
    # if the unique_constr hash is set, we know we
    # are doing an UPDATE, and we know the query
    # constraint that will be used;
    #
    # otherwise loop through all unique keys; if
    # all the columns in the key are set, then we
    # can safely use this unique key as the update
    # constraint.
    # if no update constraint can be found, this node
    # is presumed not to exist in the DB and an INSERT
    # is performed
    foreach my $uset (@usets) {
        # we already know & have the primary key
        last if %unique_constr;

        # if we are loading up a fresh/blank slate
        # database then we don't need to check for
        # existing tuples, as everything should
        # have been inserted/updated this session
        if ($self->policy_freshbulkload) {
            next;
        }

        # if an xort-style attribute has op=insert
        # this is the same as a bulkload
        if ($operation && $operation eq 'insert') {
            next;
        }

        # already tried PK
#        if (scalar(@$uset) == 1 &&
#            $uset->[0] eq $pkcol) {
#            next;
#        }
        trace(0, "TRYING USET: ;@$uset; [pk=$pkcol]") if $TRACE;

        # get the values of the unique key columns;
        # %constr is a candidate unique key=>val mapping
        my %constr =
          map {
              my $v = $node->sget($_);
              $_ => $v
          } @$uset;

        # each column in the unique key must be
        # non-NULL; try the next unique key if
        # this one is unsuitable

        # -- COMMENTED OUT cjm 20041012
        # mysql auto-creates defaults for non-null fields;
        # we cannot use this code:
        # UNCOMMENTED 20050304

        # -- make null value part of the key
        # -- ADDED 20041012 - make null 0/''

        foreach (keys %constr) {
            # in pg, pk cols are sequences with defaults nextval
            # skip these
            next if $self->is_pk_col($_);
            if (!defined($constr{$_})) {
                if ($self->is_fk_col($_)) {
                    # if xort-style, the container may be an
                    # implicit foreign key

                    # TODO: check element
                    if ($parent_pk_id) {
                        trace(0, "USING PARENT ELEMENT: $_ => $parent_pk_id");
                        $constr{$_} = $parent_pk_id;
                    }
                }
                else {
                    my $colobj = $tableobj->column($_);
                    my $default_val = $colobj->default;
                    my $col_type = $colobj->type;
                    if (defined $default_val) {
                        # problem with DBIx::DBSchema
                        if ($default_val =~ /^\'(.*)\'::/) {
                            trace(0, "FIXING DEFAULT: $default_val => $1") if $TRACE;
                            $default_val = $1;
                        }
                        if (($col_type =~ /^int/ || $col_type =~ /float/) && $default_val eq '') {
                            # this SHOULDN'T be necessary, but appears to be required for
                            # some configuartions. DBSchema problem?
                            $default_val=0;
                        }
			if (ref($default_val)) {
			    # In new versions of DBIx::DBSchema (0.38, possibly older versions),
			    # this appears to be a reference
			    $default_val = $$default_val;
			    if ($default_val eq "''") {
				$default_val = '';
			    }
			}
                        $constr{$_} = $default_val;
                        trace(0, "USING DEFAULT[type=$col_type] $_ => \"$constr{$_}\"") if $TRACE;
                    }
                }
            }
        }

        # TODO: check cases eg dbxref in chado; null default values...?
        next if grep { !defined($_) } values %constr;

        %unique_constr = %constr;
        if (!$select_col && @$uset == 1) {
            $select_col = $uset->[0];
        }
        trace(0, "GOT unique_constr, select_col=$select_col") if $TRACE;
        last;
    }
    # -- END OF @usets --

    # %unique_constr is set; a mapping for a unique key colset
    # if this is not set, then we must insert

    if (%unique_constr) {

        # -- IN-MEMORY CACHING --
        # check if we have already updated/inserted
        # this tuple this session; and if so, what
        # the update constraint used was
        if ($is_caching_on == 1 || $is_caching_on == 3) {

            #$self->throw("no select col for $element") unless $select_col;
            # fetch values of unique_constr from cache
            my %cached_colvals =
              %{$self->query_cache($element,
                                   \%unique_constr)
                  || {}};
            # have we stored anything with uniq key %unique_constr before?
            if (%cached_colvals) {
                if ($pkcol) {
                    $id = $cached_colvals{$pkcol};
                    if ($id) {
                        # use the cached pk id for efficiency
                        #%unique_constr = {$pkcol => $id};
                        trace(0, "CACHED $pkcol  = $id") if $TRACE;
                    }
                    else {
                        trace(0, "NO CACHED COLVAL FOR $pkcol :: ".
                             join("; ",map {"$_ = $cached_colvals{$_}"} keys %cached_colvals)) if $TRACE;
                    }
                }

                # yes - has it changed?
                foreach my $col (keys %cached_colvals) {
                    if ($cached_colvals{$col} && $store_hash{$col} &&
                        $cached_colvals{$col} && $store_hash{$col}) {
                        # don't bother re-storing anything
                        delete $store_hash{$col};
                    }
                }
                if (%store_hash) {
                    my @x = keys %store_hash;
                    trace(0, "WILL STORE: @x") if $TRACE;
                }
                else {
                    trace(0, "UNCHANGED - WILL NOT STORE; store_hash empty") if $TRACE;
                }
            }
            else {
            }
        }
        # -- END OF CACHING CHECK --

        # -- GET PK VAL $id BASED ON unique_constr --
        # (we may already have this based on memory-cache)
        if (!$id) {

            # the input node contains all the keys in %update_constr
            # - check to see if this relation exists in the DB

            my $vals;
            if ($is_caching_on >= 2) {
                $vals = [];
            }
            else {
                my $sql =
                  $self->makesql($element,
                                 \%unique_constr,
                                 $select_col);
                trace(0, "SQL: $sql") if $TRACE;
                $vals =
                  $dbh->selectcol_arrayref($sql);
            }

            if (@$vals) {
                # yes it does exist in DB; check if there is a
                # pkcol - if there is, it means we can do an
                # update and
                if ($pkcol && $select_col && $select_col eq $pkcol) {
                    # this is the value we return at the
                    # end
                    $id = $vals->[0];
                    if ($remap{$pkcol}) {
                        #my $colvalmap = $self->get_mapping_for_col($pkcol);
                        my $colvalmap = $self->id_remap_idx;
                        #my $colvalmap = $self->get_mapping_for_col($element);
                        $colvalmap->{$remap{$pkcol}} = $id;
                        trace(0, "COLVALMAP $pkcol $remap{$pkcol} = $id") if $TRACE;
                    }
                }
                else {
                    # $id not set, but we will later perform an update anyway
                }
            }
            else {
                # this node is not in the DB; force insert
                %unique_constr = ();
            }
        }
    } # end of get pk val

    # ---- UPDATE OR INSERT -----
    # at this stage we know if we are updating
    # or inserting, depending on whether a suitable
    # update constraint has been found

    my $this_op;
    if (%unique_constr) {
        $this_op = 'update';
    }
    else {
        $this_op = 'insert';
    }
    if (defined $operation) {
        if ($operation eq 'force') {
            $operation = $this_op;
        }
        else {
            # update/lookup/insert
            # insert: already dealt with
        }
    }
    else {
        $operation = $this_op;
    }

    if ($operation eq 'replace') {
        # replace = delete followed by insert
        if (%unique_constr) {
            $self->deleterow($element,\%unique_constr);
        }
        else {
            $self->throw("Cannot find row to delete it:\n".$node->xml);
        }
        $operation = 'insert';
    }

    if ($operation eq 'update') {
        # ** UPDATE **
        if ($self->noupdate_h->{$element}) {
            if ($tracekeyval) {
                printf STDERR "NOUPDATE: $tracenode = $tracekeyval\n"
            }
            trace(0, sprintf("NOUPDATE on %s OR child nodes (We have %s)",
                             $element,
                             join('; ',values %unique_constr)
                            )) if $TRACE;
            # don't return yet; there are still the delayed nodes
            ##return $id;
        }
        else {
            # if there are no fields modified,
            # no change
            foreach (keys %unique_constr) {
                # no point setting any column
                # that is part of the update constraint
                delete $store_hash{$_};
            }

            # only update if there are cols set that are
            # not part of unique constraint
            if (%store_hash) {
                if ($tracekeyval) {
                    printf STDERR "UPDATE: $tracenode = $tracekeyval\n"
                }

                $self->updaterow($element,
                                 \%store_hash,
                                 \%unique_constr);
                # -- CACHE RESULTS --
                if ($is_caching_on == 1 || $is_caching_on == 3) {
                    $self->update_cache($element,
                                        \%store_hash,
                                        \%unique_constr);
                }
            }
            else {
                trace(0, sprintf("NOCHANGE on %s (We have %s) id=$id",
                                 $element,
                                 join('; ',values %unique_constr)
                            )) if $TRACE;
                if ($tracekeyval) {
                    print STDERR "NOCHANGE: $tracenode = $tracekeyval\n"
                }
            }
        }
    } elsif ($operation eq 'insert') {
        # ** INSERT **
        if (%store_hash) {
            $id =
              $self->insertrow($element,
                               \%store_hash,
                               $pkcol);
            if (!$id) {
                # this only happens if $self->force(1) is set
                if (@delayed_store) {
                    print STDERR "Insert on \"$element\" did not return a primary key ID.\n Possible causes: sequence not define [Pg]?\n";
                    if ($self->force) {
                        return;
                    }
                    else {
                        confess("non-recoverable error");
                    }
                }
                return;
            }
            if ($tracekeyval) {
                printf STDERR "INSERT: $tracenode $tracekeyval [val = $id]\n"
            }
            if ($pkcol) {
                if ($remap{$pkcol}) {
                    my $colvalmap = $self->id_remap_idx;
                    #my $colvalmap = $self->get_mapping_for_col($element);
                    $colvalmap->{$remap{$pkcol}} = $id;
                    trace(0, "colvalmap $remap{$pkcol} = $id") if $TRACE;
                }
            }

            # -- CACHE RESULTS --
            if ($is_caching_on) {
                my %cache_hash = %store_hash;
                if ($pkcol) {
                    $cache_hash{$pkcol} = $id;
                }
                $self->insert_into_cache($element,
                                         \%cache_hash,
                                         \@usets);
                trace(0, "CACHING: $element") if $TRACE;
            }
        }
    }
    elsif ($operation eq 'delete') {
        if (%unique_constr) {
            $self->deleterow($element,\%unique_constr);
        }
        else {
            $self->throw("Cannot find row to delete it (perhaps unique constraint not satisfied?):\n".$node->xml);
        }
    }
    elsif ($operation eq 'lookup') {
        # lookup: do nothing, already have ID
        if (!$id) {
            $self->throw("lookup: no ID; could not find this node in db (perhaps unique constraint not satisfied?) %s:\n",$node->xml);
        }
    }
    else {
        $self->throw("cannot do op: $operation");
    } # -- end of UPDATE/INSERT/LOOKUP


    # -- DELAYED STORE --
    # Any non-terminal child nodes of the current one have
    # some kind of foreign key relationship to the current
    # relation. Either it is 1:many or many:1
    #
    # if the relation for the child node has a foreign key
    # into the current relation, we need to store the current
    # relation first to get the current relation's primary key.
    #
    # we have already done this, so now is the time to store
    # any of these child nodes
    if (@delayed_store) {
        foreach my $sn (@delayed_store) {

            my $fk; # foreign key column in subtable
            my $snname = $sn->name; # subtable name

            # if a mapping is used (eg in metadata), then
            # this takes priority
            foreach (@$mapping) {
                if ($_->name eq 'parentfk' &&
                    $_->get_table eq $snname) {
                    $fk = $_->get_col;
                }
            }

            # no mapping, by default use the current nodes primary
            # key (this assumes eg person.address_id is a fk to
            # a table with pk address_id; we will check and possibly
            # override this later)
            if (!$fk) {
                $fk = $pkcol;
            }

            # HACK!!
            # Some databases (eg GO Database) use 'id' for pk col
            # names; fks to this table will be of form _id 




2026




 





 





 





 





 





 



 	    if ($fk eq 'id') { 




2027




 





 





 





 





 





 



 		$fk = $element . '_id'; 




2028




 





 





 





 





 





 



 	    } 




2029




 





 





 





 





 





 



   




2030




 





 





 





 





 





 



             # --SET SUBNODE FK-- 




2031




 





 





 





 





 





 



             # it is necessarily true that each delayed-store subnode 




2032




 





 





 





 





 





 



             # must have some fk relationship back to the existing one 




2033




 





 





 





 





 





 



             # the subnode has a fk relation up to this one; 




2034




 





 





 





 





 





 



             # by default we assume that the subnode fk column is named 




2035




 





 





 





 





 





 



             # the same as the current pk. However, we check that this 




2036




 





 





 





 





 





 



             # is the case. If not, we deduce what the correct fk col is 




2037




 





 





 





 





 





 



             my $subtable = 




2038




 





 





 





 





 





 



               $dbschema->table($snname); 




2039




 





 





 





 





 





 



             if ($subtable->column($fk)) { 




2040




 





 





 





 





 





 



                 # a fk col with the name as the current node pk col exists; 




2041




 





 





 





 





 





 



                 # use it 




2042




 





 





 





 





 





 



   




2043




 





 





 





 





 





 



                 # do nothing - current value of $fk is fine 




2044




 





 





 





 





 





 



             } 




2045




 





 





 





 





 





 



             else { 




2046




 





 





 





 





 





 



                 # deduce actual fk column 




2047




 





 





 





 





 





 



                 # there should only be ONE subnode fk column UNSET; 




2048




 





 





 





 





 





 



                 # this implicitly refers to the current node 




2049




 





 





 





 





 





 



                 my @subcolumns = $subtable->columns; 




2050




 





 





 





 





 





 



                 my @potential_fks = (); 




2051




 





 





 





 





 





 



                 foreach my $subcolumn (@subcolumns) { 




2052




 





 





 





 





 





 



                     if ($self->is_fk_col($subcolumn) && 




2053




 





 





 





 





 





 



                         !$self->is_pk_col($subcolumn)) { 




2054




 





 





 





 





 





 



                          




2055




 





 





 





 





 





 



                         # Definite foreign key 




2056




 





 





 





 





 





 



                         if (defined $sn->sget($subcolumn)) { 




2057




 





 





 





 





 





 



                             # already set 




2058




 





 





 





 





 





 



                         } 




2059




 





 





 





 





 





 



                         else { 




2060




 





 





 





 





 





 



                             push(@potential_fks, $subcolumn); 




2061




 





 





 





 





 





 



                         } 




2062




 





 





 





 





 





 



                     } 




2063




 





 





 





 





 





 



                 } 




2064




 





 





 





 





 





 



                 trace(0, "POTENTIAL FKS: @potential_fks"); 




2065




 





 





 





 





 





 



                 if (!@potential_fks) { 




2066




 





 





 





 





 





 



                     $self->throw("I do not know what to do with the current ". 




2067




 





 





 





 





 





 



                                  "pl val ($id). There does not appear to be ". 




2068




 





 





 





 





 





 



                                  "a $fk column in $snname, and all fks in ". 




2069




 





 





 





 





 





 



                                  "the subtable $snname are currently set"); 




2070




 





 





 





 





 





 



                 } 




2071




 





 





 





 





 





 



                 if (@potential_fks > 1) { 




2072




 





 





 





 





 





 



                     $self->throw("There appear to be multiple potential fks ". 




2073




 





 





 





 





 





 



                                  "[ @potential_fks ]. I do not know which ". 




2074




 





 





 





 





 





 



                                  "to choose to assign the current pk val $id". 




2075




 





 





 





 





 





 



                                  " to"); 




2076




 





 





 





 





 





 



                 } 




2077




 





 





 





 





 





 



                 $fk = shift @potential_fks; 




2078




 





 





 





 





 





 



             } 




2079




 





 





 





 





 





 



             # -- $fk value is set 




2080




 





 





 





 





 





 



             $sn->set($fk, $id); 




2081




 





 





 





 





 





 



             # -- $fk table assigned 




2082




 





 





 





 





 





 



   




2083




 





 





 





 





 





 



             trace(0, "NOW TIME TO STORE [curr pk val = $id] [fkcol = $fk] ", $sn->xml) if $TRACE; 




2084




 





 





 





 





 





 



             # store subnode, passing in info on current node 




2085




 





 





 





 





 





 



             $self->_storenode($sn,{parent_pk_id=>$id, 




2086




 





 





 





 





 





 



                                    parent_element=>$element, 




2087




 





 





 





 





 





 



                                    assigned_node_h=>{$fk=>1}}); 




2088




 





 





 





 





 





 



         } 




2089




 





 





 





 





 





 



     } # -- end of @delayed_store 




2090




 





 





 





 





 





 



   




2091




 





 





 





 





 





 



     if ($current_attribute_node) { 




2092




 





 





 





 





 





 



         if ($id) { 




2093




 





 





 





 





 





 



             my $macro_id = $current_attribute_node->sget_id; 




2094




 





 





 





 





 





 



             if ($macro_id) { 




2095




 





 





 





 





 





 



                 $self->macro_id_h->{$macro_id} = $id; 




2096




 





 





 





 





 





 



                 trace(0, "SETTING MACRO ID MAP: $macro_id => $id") if $TRACE; 




2097




 





 





 





 





 





 



             } 




2098




 





 





 





 





 





 



             else { 




2099




 





 





 





 





 





 



             } 




2100




 





 





 





 





 





 



         } 




2101




 





 





 





 





 





 



     } 




2102




 





 





 





 





 





 



   




2103




 





 





 





 





 





 



     return $id; 




2104




 





 





 





 





 





 



 } 




2105




 





 





 





 





 





 



   




2106




 





 





 





 





 





 



 # --SQL directives embedded in XML-- 




2107




 





 





 





 





 





 



   




2108




 





 





 





 





 





 



 sub _execute_sqlnode { 




2109




 





 





 





 





 





 



     my $self = shift; 




2110




 





 





 





 





 





 



     my $sqlnode = shift; 




2111




 





 





 





 





 





 



     if ($sqlnode->element eq '_sql') { 




2112




 





 





 





 





 





 



         my $dbh = $self->dbh; 




2113




 





 





 





 





 





 



         my $op = $sqlnode->get('@/op'); 




2114




 





 





 





 





 





 



         my $col = $sqlnode->get('@/col'); 




2115




 





 





 





 





 





 



         my $table = $sqlnode->get('@/from'); 




2116




 





 





 





 





 





 



         my $match = $sqlnode->get('@/match'); 




2117




 





 





 





 





 





 



         my @subnodes = grep {$_->element ne '@'} $sqlnode->kids; 




2118




 





 





 





 





 





 



         if ($op eq 'delete') { 




2119




 





 





 





 





 





 



             my $pkey = $sqlnode->get('@/pkey'); 




2120




 





 





 





 





 





 



             trace(0,"deleting from $table"); 




2121




 





 





 





 





 





 



             my @vals = map {$self->_execute_sqlnode($_)} @subnodes; 




2122




 





 





 





 





 





 



             # do iteratively rather than in 1 SQL stmt 




2123




 





 





 





 





 





 



             if (@vals) { 




2124




 





 





 





 





 





 



                 my $sql = 




2125




 





 





 





 





 





 



                   sprintf("SELECT $pkey FROM $table WHERE $match IN (%s)", 




2126




 





 





 





 





 





 



                           join(", ",@vals)); 




2127




 





 





 





 





 





 



                 trace(0, "SQL: $sql"); 




2128




 





 





 





 





 





 



                 my $ids_to_delete =  




2129




 





 





 





 





 





 



                   $dbh->selectcol_arrayref($sql); # quote 




2130




 





 





 





 





 





 



                 foreach my $id (@$ids_to_delete) { 




2131




 





 





 





 





 





 



                     my $delete_sql = 




2132




 





 





 





 





 





 



                       "DELETE FROM $table WHERE $pkey=$id"; 




2133




 





 





 





 





 





 



                     trace(0,"SQL: $delete_sql"); 




2134




 





 





 





 





 





 



                     $dbh->do($delete_sql); 




2135




 





 





 





 





 





 



                 } 




2136




 





 





 





 





 





 



             } 




2137




 





 





 





 





 





 



         } 




2138




 





 





 





 





 





 



         elsif ($op eq "select") { 




2139




 





 





 





 





 





 



             my @vals = $sqlnode->get('.'); 




2140




 





 





 





 





 





 



             my $sql = 




2141




 





 





 





 





 





 



               sprintf("SELECT $col FROM $table WHERE $match IN (%s)", 




2142




 





 





 





 





 





 



                       join(", ",map {$dbh->quote($_)} @vals)); 




2143




 





 





 





 





 





 



             trace(0, "SQL: $sql"); 




2144




 





 





 





 





 





 



             my $ids =  




2145




 





 





 





 





 





 



               $dbh->selectcol_arrayref($sql); 




2146




 





 





 





 





 





 



             trace(0,"id list in select: @$ids"); 




2147




 





 





 





 





 





 



             return(@$ids); 




2148




 





 





 





 





 





 



         } 




2149




 





 





 





 





 





 



         else { 




2150




 





 





 





 





 





 



             $self->throw("Do not understand SQL directive: $op") 




2151




 





 





 





 





 





 



         } 




2152




 





 





 





 





 





 



     } 




2153




 





 





 





 





 





 



     else { 




2154




 





 





 





 





 





 



         return $sqlnode->data; 




2155




 





 





 





 





 





 



     } 




2156




 





 





 





 





 





 



     return; 




2157




 





 





 





 





 





 



 } 




2158




 





 





 





 





 





 



   




2159




 





 





 





 





 





 



 sub _process_sql { 




2160




 





 





 





 





 





 



     my $self = shift; 




2161




 





 





 





 





 





 



     my $node = shift; 




2162




 





 





 





 





 





 



     my $element = $node->element; 




2163




 





 





 





 





 





 



     if ($element eq 'in') { 




2164




 





 





 





 





 





 



          




2165




 





 





 





 





 





 



     } 




2166




 





 





 





 





 





 



     else { 




2167




 





 





 





 





 





 



         $self->throw("Do not understand SQL directive: $element") 




2168




 





 





 





 





 





 



     } 




2169




 





 





 





 





 





 



 } 




2170




 





 





 





 





 





 



   




2171




 





 





 





 





 





 



 # -- QUERYING -- 




2172




 





 





 





 





 





 



   




2173




 





 





 





 





 





 



 sub rmake_nesting { 




2174




 





 





 





 





 





 



     my $node = shift; 




2175




 





 





 





 





 





 



   




2176




 





 





 





 





 





 



     if ($node->element eq 'composite') { 




2177




 





 





 





 





 





 



 	my $first = $node->getnode_first; 




2178




 





 





 





 





 





 



 	my $second = $node->getnode_second; 




2179




 





 





 





 





 





 



 	my $head = rmake_nesting($first->data->[0]); 




2180




 





 





 





 





 





 



 	my $tail = rmake_nesting($second->data->[0]); 




2181




 





 





 





 





 





 



 	if ($head->isterminal) { 




2182




 





 





 





 





 





 



 	    return  




2183




 





 





 





 





 





 



 	      Data::Stag->new($head->element => [$tail]); 




2184




 





 





 





 





 





 



 	} 




2185




 





 





 





 





 





 



 	$head->addkid($tail); 




2186




 





 





 





 





 





 



 	return $head; 




2187




 





 





 





 





 





 



     } 




2188




 





 





 





 





 





 



     elsif ($node->element eq 'leaf') { 




2189




 





 





 





 





 





 



 	my $alias = $node->get_alias; 




2190




 





 





 





 





 





 



 	my $tn = $alias || $node->get_name; 




2191




 





 





 





 





 





 



 	return Data::Stag->new($tn=>1); 




2192




 





 





 





 





 





 



     } 




2193




 





 





 





 





 





 



     else { 




2194




 





 





 





 





 





 



 	die; 




2195




 





 





 





 





 





 



     } 




2196




 





 





 





 





 





 



 } 




2197




 





 





 





 





 





 



   




2198




 





 





 





 





 





 



 # if true, a metadata tag will be added to stag nodes selected from db 




2199




 





 





 





 





 





 



 sub include_metadata { 




2200




 





 





 





 





 





 



     my $self = shift; 




2201




 





 





 





 





 





 



     $self->{_include_metadata} = shift if @_; 




2202




 





 





 





 





 





 



     return $self->{_include_metadata}; 




2203




 





 





 





 





 





 



 } 




2204




 





 





 





 





 





 



   




2205




 





 





 





 





 





 



   




2206




 





 





 





 





 





 



 # last SQL SELECT statement executed 




2207




 





 





 





 





 





 



 sub last_stmt { 




2208




 





 





 





 





 





 



     my $self = shift; 




2209




 





 





 





 





 





 



     $self->{_last_stmt} = shift if @_; 




2210




 





 





 





 





 





 



     return $self->{_last_stmt}; 




2211




 





 





 





 





 





 



 } 




2212




 





 





 





 





 





 



   




2213




 





 





 





 





 





 



 sub last_sql_and_args { 




2214




 





 





 





 





 





 



     my $self = shift; 




2215




 





 





 





 





 





 



     $self->{_last_sql_and_args} = shift if @_; 




2216




 





 





 





 





 





 



     return $self->{_last_sql_and_args}; 




2217




 





 





 





 





 





 



 } 




2218




 





 





 





 





 





 



   




2219




 





 





 





 





 





 



   




2220




 





 





 





 





 





 



 sub sax_handler { 




2221




 





 





 





 





 





 



     my $self = shift; 




2222




 





 





 





 





 





 



     $self->{_sax_handler} = shift if @_; 




2223




 





 





 





 





 





 



     return $self->{_sax_handler}; 




2224




 





 





 





 





 





 



 } 




2225




 





 





 





 





 





 



   




2226




 





 





 





 





 





 



   




2227




 





 





 





 





 





 



 # delegates to selectall_stag and turns tree to XML 




2228




 





 





 





 





 





 



 sub selectall_xml { 




2229




 





 





 





 





 





 



     my $self = shift; 




2230




 





 





 





 





 





 



     my $stag = $self->selectall_stag(@_); 




2231




 





 





 





 





 





 



     return $stag->xml; 




2232




 





 





 





 





 





 



 } 




2233




 





 





 





 





 





 



   




2234




 





 





 





 





 





 



 # delegates to selectall_stag and turns tree to SAX 




2235




 





 





 





 





 





 



 # (candidate for optimisation - TODO - use event firing model) 




2236




 





 





 





 





 





 



 sub selectall_sax { 




2237




 





 





 





 





 





 



     my $self = shift; 




2238




 





 





 





 





 





 



     my ($sql, $nesting, $h) =  




2239




 





 





 





 





 





 



       rearrange([qw(sql nesting handler)], @_); 




2240




 





 





 





 





 





 



     my $stag = $self->selectall_stag(@_); 




2241




 





 





 





 





 





 



     $h = $h || $self->sax_handler; 




2242




 





 





 





 





 





 



     if (!$h) { 




2243




 





 





 





 





 





 



 	$self->throw("You must specify the sax handler;\n". 




2244




 





 





 





 





 





 



 		     "Either use \$dbh->sax_handler(\$h), or \n". 




2245




 





 





 





 





 





 



 		     "\$dbh->selectall_sax(-sql=>\$sql, handler->\$h)"); 




2246




 





 





 





 





 





 



     } 




2247




 





 





 





 





 





 



     return $stag->sax($h); 




2248




 





 





 





 





 





 



 } 




2249




 





 





 





 





 





 



   




2250




 





 





 





 





 





 



 # delegates to selectall_stag and turns tree to S-Expression 




2251




 





 





 





 





 





 



 sub selectall_sxpr { 




2252




 





 





 





 





 





 



     my $self = shift; 




2253




 





 





 





 





 





 



     my $stag = $self->selectall_stag(@_); 




2254




 





 





 





 





 





 



     return $stag->sxpr; 




2255




 





 





 





 





 





 



 } 




2256




 





 





 





 





 





 



   




2257




 





 





 





 





 





 



 # does not bother decomposing and nesting the results; just 




2258




 





 





 





 





 





 



 # returns the denormalised table from the SQL query. 




2259




 





 





 





 





 





 



 # arrayref of arrayrefs - rows x cols 




2260




 





 





 





 





 





 



 # first row of rows is column headings 




2261




 





 





 





 





 





 



 sub selectall_rows { 




2262




 





 





 





 





 





 



     my $self = shift; 




2263




 





 





 





 





 





 



     my ($sql, $nesting, $bind, $template) =  




2264




 





 





 





 





 





 



       rearrange([qw(sql nesting bind template)], @_); 




2265




 





 





 





 





 





 



     my $rows = 




2266




 





 





 





 





 





 



       $self->selectall_stag(-sql=>$sql, 




2267




 





 





 





 





 





 



 			    -nesting=>$nesting, 




2268




 





 





 





 





 





 



 			    -bind=>$bind, 




2269




 





 





 





 





 





 



 			    -template=>$template, 




2270




 





 





 





 





 





 



 			    -return_arrayref=>1, 




2271




 





 





 





 





 





 



 			   ); 




2272




 





 





 





 





 





 



     return $rows; 




2273




 





 





 





 





 





 



 } 




2274




 





 





 





 





 





 



   




2275




 





 





 





 





 





 



 # --------------------------------------- 




2276




 





 





 





 





 





 



 # selectall_stag(sql, nesting) 




2277




 





 





 





 





 





 



 # 




2278




 





 





 





 





 





 



 # Takes an sql string containing a SELECT statement, 




2279




 





 





 





 





 





 



 # parses it to get the tree structure; this can be 




2280




 





 





 





 





 





 



 # overridden with the nesting optional argument. 




2281




 





 





 





 





 





 



 # 




2282




 





 





 





 





 





 



 # The SELECT statement is executed, and the relations are 




2283




 





 





 





 





 





 



 # transformed into a stag tree 




2284




 





 





 





 





 





 



 # 




2285




 





 





 





 





 





 



 # --------------------------------------- 




2286




 





 





 





 





 





 



 sub selectall_stag { 




2287




 





 





 





 





 





 



     my $self = shift; 




2288




 





 





 





 





 





 



     my ($sql, $nesting, $bind, $template, $return_arrayref, $include_metadata, $aliaspolicy) =  




2289




 





 





 





 





 





 



       rearrange([qw(sql nesting bind template return_arrayref include_metadata aliaspolicy)], @_); 




2290




 





 





 





 





 





 



     my $prep_h = $self->prepare_stag(@_); 




2291




 





 





 





 





 





 



     my $cols = $prep_h->{cols}; 




2292




 





 





 





 





 





 



     my $sth = $prep_h->{sth}; 




2293




 





 





 





 





 





 



     my $exec_args = $prep_h->{exec_args}; 




2294




 





 





 





 





 





 



   




2295




 





 





 





 





 





 



     if (!defined($include_metadata)) { 




2296




 





 





 





 





 





 



         $include_metadata = $self->include_metadata; 




2297




 





 





 





 





 





 



     } 




2298




 





 





 





 





 





 



   




2299




 





 





 





 





 





 



     # TODO - make this event based so we don't have to 




2300




 





 





 





 





 





 



     # load all into memory 




2301




 





 





 





 





 





 



     my $rows = 




2302




 





 





 





 





 





 



       $self->dbh->selectall_arrayref($sth, undef, @$exec_args); 




2303




 





 





 





 





 





 



     if ($return_arrayref) { 




2304




 





 





 





 





 





 



 	my @hdrs = (); 




2305




 





 





 





 





 





 



 	for (my $i=0; $i<@$cols; $i++) { 




2306




 





 





 





 





 





 



 	    my $h = $prep_h->{col_aliases_ordered}->[$i] || $cols->[$i]; 




2307




 





 





 





 





 





 



 	    push(@hdrs, $h); 




2308




 





 





 





 





 





 



 	} 




2309




 





 





 





 





 





 



 	return [\@hdrs, @$rows]; 




2310




 





 





 





 





 





 



     } 




2311




 





 





 





 





 





 



   




2312




 





 





 





 





 





 



     trace(0, sprintf("Got %d rows\n", scalar(@$rows))) if $TRACE; 




2313




 





 





 





 





 





 



     # --- reconstruct tree from relations 




2314




 





 





 





 





 





 



     my $stag = 




2315




 





 





 





 





 





 



       $self->reconstruct( 




2316




 





 





 





 





 





 



                          -rows=>$rows, 




2317




 





 





 





 





 





 



                          -cols=>$cols, 




2318




 





 





 





 





 





 



                          -alias=>$prep_h->{alias}, 




2319




 





 





 





 





 





 



                          -nesting=>$prep_h->{nesting}, 




2320




 





 





 





 





 





 



                          -aliaspolicy=>$aliaspolicy, 




2321




 





 





 





 





 





 



                         ); 




2322




 





 





 





 





 





 



     if ($include_metadata) { 




2323




 





 





 





 





 





 



         my ($last_sql, @sql_args) = @{$self->last_sql_and_args || []}; 




2324




 





 





 





 





 





 



         my @kids = $stag->kids; 




2325




 





 





 





 





 





 



         my @bind_nodes; 




2326




 





 





 





 





 





 



         if ($bind && ref($bind) eq 'HASH') { 




2327




 





 





 





 





 





 



             @bind_nodes = (stag_unflatten(argset=>[%$bind])); 




2328




 





 





 





 





 





 



         } 




2329




 





 





 





 





 





 



         unshift(@kids, 




2330




 





 





 





 





 





 



                 [dbstag_metadata=>[ 




2331




 





 





 





 





 





 



                                    [sql=>$last_sql], 




2332




 





 





 





 





 





 



                                    [nesting=>$nesting], 




2333




 





 





 





 





 





 



                                    [template=>$template], 




2334




 





 





 





 





 





 



                                    @bind_nodes, 




2335




 





 





 





 





 





 



                                    (map {[exec_arg=>$_]} @sql_args) 




2336




 





 





 





 





 





 



                                   ]]); 




2337




 





 





 





 





 





 



         $stag->kids(@kids); 




2338




 





 





 





 





 





 



     } 




2339




 





 





 





 





 





 



     return $stag; 




2340




 





 





 





 





 





 



 } 




2341




 





 





 





 





 





 



   




2342




 





 





 





 





 





 



 sub prepare_stag { 




2343




 





 





 





 





 





 



     my $self = shift; 




2344




 





 





 





 





 





 



     my ($sql, $nesting, $bind, $template, $return_arrayref, $aliaspolicy) =  




2345




 





 





 





 





 





 



       rearrange([qw(sql nesting bind template return_arrayref aliaspolicy)], @_); 




2346




 





 





 





 





 





 



   




2347




 





 





 





 





 





 



     my $parser = $self->parser; 




2348




 





 





 





 





 





 



   




2349




 





 





 





 





 





 



     my $sth; 




2350




 





 





 





 





 





 



     my @exec_args = (); 




2351




 





 





 





 





 





 



     if (ref($sql)) { 




2352




 





 





 





 





 





 



 	$template = $sql; 




2353




 





 





 





 





 





 



     } 




2354




 





 





 





 





 





 



     if ($template) { 




2355




 





 





 





 





 





 



 	if (!ref($template)) { 




2356




 





 





 





 





 





 



 	    $template = $self->find_template($template); 




2357




 





 





 





 





 





 



 	} 




2358




 





 





 





 





 





 



 	($sql, @exec_args) = $template->get_sql_and_args($bind); 




2359




 





 





 





 





 





 



     } 




2360




 





 





 





 





 





 



     trace 0, "parsing_sql: $sql\n"; 




2361




 





 





 





 





 





 



   




2362




 





 





 





 





 





 



     # PRE-parse SQL statement for stag-specific extensions 




2363




 





 





 





 





 





 



     if ($sql =~ /(.*)\s+use\s+nesting\s*(.*)/si) { 




2364




 





 





 





 





 





 



 	my ($pre, $post) = ($1, $2); 




2365




 





 





 





 





 





 



 	my ($extracted, $remainder) = 




2366




 





 





 





 





 





 



 	  extract_bracketed($post, '()'); 




2367




 





 





 





 





 





 



         if ($nesting) { 




2368




 





 





 





 





 





 



             $self->throw("nestings clash: $nesting vs $extracted"); 




2369




 





 





 





 





 





 



         } 




2370




 





 





 





 





 





 



 	$nesting = Data::Stag->parsestr($extracted); 




2371




 





 





 





 





 





 



 	$sql = "$pre $remainder"; 




2372




 





 





 





 





 





 



     } 




2373




 





 





 





 





 





 



   




2374




 





 





 





 





 





 



   




2375




 





 





 





 





 





 



     # get the parsed SQL SELECT statement as a stag node 




2376




 





 





 





 





 





 



     my $stmt = $parser->selectstmt($sql); 




2377




 





 





 





 





 





 



     if (!$stmt) { 




2378




 





 





 





 





 





 



 	# there was some error parsing the SQL; 




2379




 





 





 





 





 





 



 	# DBI can probably give a better explanation. 




2380




 





 





 





 





 





 



 	eval { 




2381




 





 





 





 





 





 



 	    my $sth = $self->dbh->prepare($sql); 




2382




 





 





 





 





 





 



 	     




2383




 





 





 





 





 





 



 	}; 




2384




 





 





 





 





 





 



 	if ($@) { 




2385




 





 





 





 





 





 



 	    $self->throw("SQL ERROR:\n$@"); 




2386




 





 





 





 





 





 



 	} 




2387




 





 





 





 





 





 



 	# DBI accepted it - must be a bug in the DBStag grammar 




2388




 





 





 





 





 





 



 	$self->throw("I'm sorry but the SQL statement you gave does\n". 




2389




 





 





 





 





 





 



 		     "not conform to the more limited subset of SQL\n". 




2390




 





 





 





 





 





 



 		     "that DBStag supports. Please see the DBStag docs\n". 




2391




 





 





 





 





 





 



 		     "for details.\n". 




2392




 





 





 





 





 





 



 		     "\n". 




2393




 





 





 





 





 





 



 		     "Remember to check you explicitly declare all aliases\n". 




2394




 





 





 





 





 





 



 		     "using AS\n\n\nSQL:$sql"); 




2395




 





 





 





 





 





 



     } 




2396




 





 





 





 





 





 



   




2397




 





 





 





 





 





 



   




2398




 





 





 





 





 





 



     trace 0, "parsed_sql: $sql\n"; 




2399




 





 





 





 





 





 



 #    trace 0, $stmt->xml; 




2400




 





 





 





 





 





 



     my $dbschema = $self->dbschema; 




2401




 





 





 





 





 





 



   




2402




 





 





 





 





 





 



     $self->last_stmt($stmt); 




2403




 





 





 





 





 





 



   




2404




 





 





 





 





 





 



     # stag node of FROM part of SQL 




2405




 





 





 





 





 





 



     my $fromstruct = $stmt->get_from; 




2406




 





 





 





 





 





 



   




2407




 





 





 





 





 





 



     # --- aliases --- 




2408




 





 





 





 





 





 



   




2409




 





 





 





 





 





 



     # keep a hash of table aliases 




2410




 





 





 





 





 





 



     # KEY: table alias 




2411




 





 





 





 





 





 



     # VAL: base table 




2412




 





 





 





 





 





 



     # for example, 'SELECT * FROM person AS p' 




2413




 





 





 





 





 





 



     # will result in $alias_h = { p => person } 




2414




 





 





 





 





 





 



     my $alias_h = {}; 




2415




 





 





 





 





 





 



   




2416




 





 





 





 





 





 



     # build alias hash using FROM node 




2417




 





 





 





 





 





 



     foreach my $sn ($fromstruct->subnodes) { 




2418




 





 





 





 





 





 



         get_table_alias_map($sn, $alias_h); 




2419




 





 





 





 





 





 



     } 




2420




 





 





 





 





 





 



   




2421




 





 





 





 





 





 



     # as well as an alias hash map,  




2422




 





 





 





 





 





 



     # keep an array of stag nodes representing all the aliases 




2423




 





 





 





 





 





 



     my @aliases = (); 




2424




 





 





 





 





 





 



     foreach my $alias (keys %$alias_h) { 




2425




 





 





 





 





 





 



         push(@aliases,  




2426




 





 





 





 





 





 



              Data::Stag->new(alias=>[ 




2427




 





 





 





 





 





 



                                      [name=>$alias], 




2428




 





 





 





 





 





 



                                      [table=>$alias_h->{$alias}->[0]] 




2429




 





 





 





 





 





 



                                     ])); 




2430




 





 





 





 





 





 



     } 




2431




 





 





 





 





 





 



     my $aliasstruct = Data::Stag->new(alias=>[@aliases]); 




2432




 





 





 





 





 





 



   




2433




 





 





 





 





 





 



     # --- nestings --- 




2434




 





 





 





 





 





 



     # 




2435




 





 





 





 





 





 



     # the cartesian product that results from a SELECT can 




2436




 





 





 





 





 





 



     # be turned into a tree - there is more than one tree to 




2437




 





 





 





 





 





 



     # choose from; eg with "x NJ y NJ z" we can have trees: 




2438




 





 





 





 





 





 



     # [x [y [z]]] 




2439




 





 





 





 





 





 



     # [x [y z]] 




2440




 





 





 





 





 





 



     # [z [x y]] 




2441




 





 





 





 





 





 



     # etc 




2442




 





 





 





 





 





 



     # 




2443




 





 





 





 





 





 



     # the actual allowed nestings through the graph is constrained 




2444




 





 





 





 





 





 



     # by the FK relationships; we do not utilise this yet (TODO!) 




2445




 





 





 





 





 





 



     # later the user need only specify the root. for now they 




2446




 





 





 





 





 





 



     # must specify the full nesting OR allow the bracket structure 




2447




 





 





 





 





 





 



     # of the joins... 




2448




 





 





 





 





 





 



   




2449




 





 





 





 





 





 



     # if the user did not explicitly supply a nesting, 




2450




 





 





 





 





 





 



     # guess one from the bracket structure of the FROM  




2451




 





 





 





 





 





 



     # clause (see rmake_nesting) 




2452




 





 





 





 





 





 



     # [TODO: be more clever in guessing the nesting using FKs] 




2453




 





 





 





 





 





 



     if (!$nesting) { 




2454




 





 





 





 





 





 



 	$nesting = Data::Stag->new(top=>1); 




2455




 





 





 





 





 





 



 #	my $cons = rmake_cons($fromstruct->data->[0], $nesting); 




2456




 





 





 





 





 





 



 	$nesting = rmake_nesting($fromstruct->data->[0]); 




2457




 





 





 





 





 





 



         $nesting = Data::Stag->new(top=>[$nesting]); 




2458




 





 





 





 





 





 



 	trace(0, "\n\nNesting:\n%s\n\n",  $nesting->xml) if $TRACE; 




2459




 





 





 





 





 





 



     } 




2460




 





 





 





 





 





 



     if ($nesting && !ref($nesting)) { 




2461




 





 





 





 





 





 



         $nesting = Data::Stag->parsestr($nesting); 




2462




 





 





 





 





 





 



     } 




2463




 





 





 





 





 





 



   




2464




 





 





 





 





 





 



     # keep an array of named relations used in the query - 




2465




 





 





 





 





 





 



     # the named relation is the alias if present; 




2466




 





 





 





 





 





 



     # eg 




2467




 





 





 





 





 





 



     # SELECT * FROM person AS p NATURAL JOIN department 




2468




 





 





 





 





 





 



     # the named relations here are 'p' and 'department' 




2469




 





 





 





 





 





 



     my @namedrelations = (); 




2470




 





 





 





 





 





 



     $fromstruct->iterate(sub { 




2471




 





 





 





 





 





 



 			     my $n = shift; 




2472




 





 





 





 





 





 



 			     if ($n->element eq 'leaf') { 




2473




 





 





 





 





 





 



                                  my $v = $n->sget_alias || $n->sget_name; 




2474




 





 





 





 





 





 



 				 push(@namedrelations, $v) 




2475




 





 





 





 





 





 



 			     } 




2476




 





 





 





 





 





 



 			 }); 




2477




 





 





 





 





 





 



   




2478




 





 





 





 





 





 



     # --- fetch columns --- 




2479




 





 





 





 





 





 



     # 




2480




 





 





 





 





 





 



     # loop through all the columns in the SELECT clause 




2481




 





 





 





 





 





 



     # making them all of a standard form; eg dealing 




2482




 





 





 





 





 





 



     # with functions and '*' wildcards appropriately 




2483




 





 





 





 





 





 



   




2484




 





 





 





 





 





 



     my @col_aliases_ordered = (); 




2485




 





 





 





 





 





 



     my @cols = 




2486




 





 





 





 





 





 



       map { 




2487




 





 





 





 





 





 



 	  # $_ iterator variable is over the columns 




2488




 





 





 





 





 





 



 	  # specified in the SELECT part of the query; 




2489




 





 





 





 





 





 



 	  # each column is represented as a stag node 




2490




 





 





 





 





 





 



   




2491




 





 





 





 





 





 



 	  # column name 




2492




 





 





 





 





 





 



           my $name = $_->get_name; 




2493




 





 





 





 





 





 



   




2494




 





 





 





 





 





 



 	  # column alias, if exists 




2495




 





 





 





 





 





 



 	  # eg in 'SELECT name AS n' the alias is 'n' 




2496




 





 





 





 





 





 



           my $col_alias = $_->get_alias; 




2497




 





 





 





 





 





 



 	  push(@col_aliases_ordered, $col_alias); 




2498




 





 





 





 





 





 



   




2499




 





 





 





 





 





 



 	  # make the name the alias; prepend named relation if supplied. 




2500




 





 





 





 





 





 



 	  # eg in 'SELECT person.name AS n' the name will become 




2501




 





 





 





 





 





 



 	  # 'person.n' 




2502




 





 





 





 





 





 



 	  if ($col_alias) { 




2503




 





 





 





 





 





 



 	      $name = $col_alias; 




2504




 





 





 





 





 





 



 	      if ($_->get_table) { 




2505




 





 





 





 





 





 



 		  $name = $_->get_table . '.'. $name; 




2506




 





 





 





 





 





 



 	      } 




2507




 





 





 





 





 





 



 	  } 




2508




 





 





 





 





 





 



   




2509




 





 





 





 





 





 



 	  my $func = $_->getnode('func'); 




2510




 





 





 





 





 





 



   




2511




 





 





 





 





 





 



 	  # from here on determines returned value of the 




2512




 





 





 





 





 





 



 	  # map iteration: 




2513




 





 





 





 





 





 



   




2514




 





 





 





 





 





 



 	  if ($func) { 




2515




 





 





 





 





 





 



 	      # a typical column node for a function looks like 




2516




 





 





 





 





 





 



 	      # this: 




2517




 





 





 





 





 





 



 	      # 




2518




 





 





 





 





 





 



 	      #    (col 




2519




 





 





 





 





 





 



 	      #      (func 




2520




 





 





 





 





 





 



 	      #        (name "somefunc") 




2521




 





 





 





 





 





 



 	      #        (args 




2522




 





 





 





 





 





 



 	      #          (col 




2523




 





 





 





 





 





 



 	      #            (name "x.foo") 




2524




 





 





 





 





 





 



 	      #            (table "x"))))) 




2525




 





 





 





 





 





 



 	      #      (alias "myname")) 




2526




 





 





 





 





 





 



 	      # 




2527




 





 





 





 





 





 



 	      # if a function is included, and the function 




2528




 





 





 





 





 





 



 	      # return value is aliased, use that alias; 




2529




 





 





 





 





 





 



 	      # otherwise ... 




2530




 





 





 





 





 





 



   




2531




 





 





 





 





 





 



 	      my $funcname = $func->get_name; 




2532




 





 





 





 





 





 



 	      # query the function stag node for the element 




2533




 





 





 





 





 





 



 	      # 'col' 




2534




 





 





 





 





 





 



 	      my ($col) =  




2535




 





 





 





 





 





 



 		$func->where('col', 




2536




 





 





 





 





 





 



 			     sub {shift->get_table}); 




2537




 





 





 





 





 





 



 	      my $table = $col_alias || $funcname; 




2538




 





 





 





 





 





 



 	      if (!$col_alias) { 




2539




 





 





 





 





 





 



 		  $col_alias = $funcname; 




2540




 





 





 





 





 





 



 	      } 




2541




 





 





 





 





 





 



 	      if ($col) { 




2542




 





 





 





 





 





 



 		  $table = $col->get_table; 




2543




 





 





 





 





 





 



 	      } 




2544




 





 





 





 





 





 



 #	      if ($col_alias =~ /(\w+)__(\w+)/) { 




2545




 





 





 





 





 





 



 #		  $table = $1; 




2546




 





 





 





 





 





 



 #		  $col_alias = $2; 




2547




 





 





 





 





 





 



 #	      } 




2548




 





 





 





 





 





 



 	      $name = $table . '.' . $col_alias; 




2549




 





 





 





 





 





 



 	      # return: 




2550




 





 





 





 





 





 



 	      $name; 




2551




 





 





 





 





 





 



 	  } 




2552




 





 





 





 





 





 



           elsif ($name =~ /^(\w+)\.\*$/) { 




2553




 





 





 





 





 





 



 	      # if the column name is of the form 




2554




 





 





 





 





 





 



 	      # RELATION.*, then replace the * with 




2555




 





 





 





 





 





 



 	      # all the actual columns from the base relation 




2556




 





 





 





 





 





 



 	      # RELATION 




2557




 





 





 





 





 





 



 	      # 




2558




 





 





 





 





 





 



 	      # the final result will be TABLE.col1, TABLE.col2,... 




2559




 





 





 





 





 





 



   




2560




 





 





 





 





 





 



               my $tn = $1; 




2561




 





 





 





 





 





 



               my $tn_alias = $tn; 




2562




 





 





 





 





 





 



   




2563




 





 





 





 





 





 



 	      # use base relation name to introspect schema 




2564




 





 





 





 





 





 



               if ($alias_h->{$tn}) { 




2565




 





 





 





 





 





 



                   $tn = $alias_h->{$tn}->[0]; 




2566




 





 





 





 





 





 



               } 




2567




 





 





 





 





 





 



               my $tbl = $dbschema->table(lc($tn)); 




2568




 





 





 





 





 





 



               if (!$tbl) { 




2569




 





 





 





 





 





 



                   confess("No such table as $tn"); 




2570




 





 





 





 





 





 



               } 




2571




 





 





 





 





 





 



 	      # introspect schema to get columns for this table 




2572




 





 





 





 





 





 



               my @cns = $tbl->columns; 




2573




 





 





 





 





 





 



   




2574




 





 





 





 





 





 



 #	      trace(0, Dumper $tbl) if $TRACE; 




2575




 





 





 





 





 





 



 	      trace(0, "TN:$tn ALIAS:$tn_alias COLS:@cns") if $TRACE; 




2576




 





 





 





 





 





 



   




2577




 





 





 





 





 





 



 	      # return: 




2578




 





 





 





 





 





 



               map { "$tn_alias.$_" } @cns; 




2579




 





 





 





 





 





 



           } 




2580




 





 





 





 





 





 



           elsif ($name =~ /^\*$/) { 




2581




 





 





 





 





 





 



 	      # if the column name is '*' (ie select all) 




2582




 





 





 





 





 





 



 	      # then replace the * with 




2583




 





 





 





 





 





 



 	      # all the actual columns from the base relations in 




2584




 





 





 





 





 





 



 	      # the query (use FROM clause) 




2585




 





 





 





 





 





 



 	      # 




2586




 





 





 





 





 





 



   




2587




 





 





 





 





 





 



               my %got = (); 




2588




 





 





 





 





 





 



               my @allcols = 




2589




 





 





 





 





 





 



                 map { 




2590




 





 





 





 





 





 



                     my $tn = $_; 




2591




 





 





 





 





 





 



                     my $baserelname =  




2592




 





 





 





 





 





 



                       $alias_h->{$tn} ?  




2593




 





 





 





 





 





 



                         $alias_h->{$tn}->[0] : $tn; 




2594




 





 





 





 





 





 



                     my $tbl = $dbschema->table(lc($baserelname)); 




2595




 





 





 





 





 





 



                     if (!$tbl) { 




2596




 





 





 





 





 





 



                         confess("Don't know anything about table:$tn\n". 




2597




 





 





 





 





 





 



                                 "Maybe DBIx::DBSchema does not work for your DBMS?\n". 




2598




 





 





 





 





 





 



                                 "If $tn is a view, you may need to modify DBIxLLDBSchema"); 




2599




 





 





 





 





 





 



                     } 




2600




 





 





 





 





 





 



                     my @cns = $tbl->columns; 




2601




 





 





 





 





 





 



                     #                  @cns = grep { !$got{$_}++ } @cns; 




2602




 





 





 





 





 





 



                     map { "$tn.$_"} @cns; 




2603




 





 





 





 





 





 



                 } @namedrelations; 




2604




 





 





 





 





 





 



   




2605




 





 





 





 





 





 



               # This is a bit hacky; if the user specifies 




2606




 





 





 





 





 





 



               # SELECT * FROM... then there is no way 




2607




 





 





 





 





 





 



               # to introspect the actual column returned 




2608




 





 





 





 





 





 



               # using DBI->selectall_arrayref 




2609




 





 





 





 





 





 



               # 




2610




 





 





 





 





 





 



               # maybe we should selectall_hashref 




2611




 





 





 





 





 





 



               # instead? this is generally slower; also 




2612




 





 





 





 





 





 



               # even if we get it with a hashref, the 




2613




 





 





 





 





 





 



               # result can be ambiguous since DBI only 




2614




 





 





 





 





 





 



               # gives us the colun names back 




2615




 





 





 





 





 





 



               # 




2616




 





 





 





 





 





 



               # to get round this we just replace the * 




2617




 





 





 





 





 





 



               # in the user's query (ie in the actual SQL) 




2618




 





 





 





 





 





 



 	      # with the full column list 




2619




 





 





 





 





 





 



               my $replace = join(', ', @allcols); 




2620




 





 





 





 





 





 



 	      # rewrite SQL statement; assum only one instance of 




2621




 





 





 





 





 





 



 	      # string '*' in these cases 




2622




 





 





 





 





 





 



               $sql =~ s/\*/$replace/; 




2623




 





 





 





 





 





 



 	      # return: 




2624




 





 





 





 





 





 



               @allcols; 




2625




 





 





 





 





 





 



 	  } 




2626




 





 





 





 





 





 



           else { 




2627




 





 





 





 





 





 



 	      # no * wildcard in column, and not a function; 




2628




 





 





 





 





 





 



 	      # just give back the node 




2629




 





 





 





 





 





 



   




2630




 





 





 





 





 





 



 	      # return: 




2631




 





 





 





 





 





 



               $name 




2632




 





 





 





 





 





 



           } 




2633




 





 





 





 





 





 



       } $stmt->sgetnode_cols->getnode_col; 




2634




 





 





 





 





 





 



   




2635




 





 





 





 





 





 



     @cols = 




2636




 





 





 





 





 





 



       map { 




2637




 





 





 





 





 





 



 	  if (/(\w+)__(\w+)/) { 




2638




 





 





 





 





 





 



 	      "$1.$2"; 




2639




 





 





 





 





 





 



 	  } 




2640




 





 





 





 





 





 



 	  else { 




2641




 





 





 





 





 





 



 	      $_ 




2642




 





 





 





 





 





 



 	  } 




2643




 





 





 





 





 





 



       } @cols; 




2644




 





 





 





 





 





 



   




2645




 





 





 





 





 





 



     # ---- end of column fetching --- 




2646




 





 





 





 





 





 



   




2647




 





 





 





 





 





 



     trace(0, "COLS:@cols") if $TRACE; 




2648




 





 





 





 





 





 



   




2649




 





 





 





 





 





 



   




2650




 





 





 





 





 





 



   




2651




 





 





 





 





 





 



     # --- execute SQL SELECT statement --- 




2652




 





 





 





 





 





 



     if ($template) { 




2653




 





 





 





 





 





 



 	$sth = $template->cached_sth->{$sql}; 




2654




 





 





 





 





 





 



 	if (!$sth) { 




2655




 





 





 





 





 





 



 	    $sth = $self->dbh->prepare($sql); 




2656




 





 





 





 





 





 



 	    $template->cached_sth->{$sql} = $sth;	   




2657




 





 





 





 





 





 



 	} 




2658




 





 





 





 





 





 



 #	($sql, $sth, @exec_args) =  




2659




 





 





 





 





 





 



 #	  $template->prepare($self->dbh, $bind); 




2660




 





 





 





 





 





 



     } 




2661




 





 





 





 





 





 



     else { 




2662




 





 





 





 





 





 



         $sth = $self->dbh->prepare($sql); 




2663




 





 





 





 





 





 



     } 




2664




 





 





 





 





 





 



     my $sql_or_sth = $sql; 




2665




 





 





 





 





 





 



     if ($sth) { 




2666




 





 





 





 





 





 



 	$sql_or_sth = $sth; 




2667




 





 





 





 





 





 



     } 




2668




 





 





 





 





 





 



     trace(0, "SQL:$sql") if $TRACE; 




2669




 





 





 





 





 





 



     trace(0, "Exec_args: @exec_args") if $TRACE && @exec_args; 




2670




 





 





 





 





 





 



     $self->last_sql_and_args([$sql, @exec_args]); 




2671




 





 





 





 





 





 



     return  




2672




 





 





 





 





 





 



       { 




2673




 





 





 





 





 





 



        sth=>$sth, 




2674




 





 





 





 





 





 



        exec_args=>\@exec_args, 




2675




 





 





 





 





 





 



        cols=>\@cols, 




2676




 





 





 





 





 





 



        col_aliases_ordered=>\@col_aliases_ordered, 




2677




 





 





 





 





 





 



        alias=>$aliasstruct, 




2678




 





 





 





 





 





 



        nesting=>$nesting 




2679




 





 





 





 





 





 



       }; 




2680




 





 





 





 





 





 



 } 




2681




 





 





 





 





 





 



   




2682




 





 





 





 





 





 



   




2683




 





 





 





 





 





 



 # ============================ 




2684




 





 





 





 





 





 



 # get_table_alias_map(tablenode, alias hash) 




2685




 





 





 





 





 





 



 # 




2686




 





 





 





 





 





 



 # checks a tablenode (eg the stag representing  




2687




 





 





 





 





 





 



 # a table construct in the FROM clause) and adds 




2688




 





 





 





 





 





 



 # it to the alias hash if it specifies an alias 




2689




 





 





 





 





 





 



 # ============================ 




2690




 





 





 





 





 





 



 sub get_table_alias_map { 




2691




 





 





 





 





 





 



     my $s = shift; 




2692




 





 





 





 





 





 



     my $h = shift; 




2693




 





 





 





 





 





 



   




2694




 





 





 





 





 





 



     # the FROM clause is natively stored as a binary tree 




2695




 





 





 





 





 





 



     # (in order to group the joins by brackets) - recursively 




2696




 





 





 





 





 





 



     # descend building the hash map 




2697




 





 





 





 





 





 



   




2698




 





 





 





 





 





 



     if ($s->name eq 'leaf') { 




2699




 





 





 





 





 





 



         my $alias = $s->get_alias; 




2700




 





 





 





 





 





 



         if ($alias) { 




2701




 





 





 





 





 





 



             $h->{$alias} = [$s->get_name]; 




2702




 





 





 





 





 





 



         } 




2703




 





 





 





 





 





 



         return ($s->get_name); 




2704




 





 





 





 





 





 



     } 




2705




 





 





 





 





 





 



     elsif ($s->name eq 'composite') { 




2706




 





 





 





 





 





 



         my ($first, $second) =  




2707




 





 





 





 





 





 



           ($s->getnode_first, 




2708




 





 





 





 





 





 



            $s->getnode_second); 




2709




 





 





 





 





 





 



         my $alias = $s->get_alias; 




2710




 





 





 





 





 





 



         my @sn = ($first->subnodes, $second->subnodes); 




2711




 





 





 





 





 





 



         my @subtbls = map { 




2712




 





 





 





 





 





 



             get_table_alias_map($_, $h), 




2713




 





 





 





 





 





 



         } @sn; 




2714




 





 





 





 





 





 



         if ($alias) { 




2715




 





 





 





 





 





 



             $h->{$alias} = [@subtbls]; 




2716




 





 





 





 





 





 



         } 




2717




 





 





 





 





 





 



         return @subtbls; 




2718




 





 





 





 





 





 



     } 




2719




 





 





 





 





 





 



     else { 




2720




 





 





 





 





 





 



         confess $s->name; 




2721




 





 





 





 





 





 



     } 




2722




 





 





 





 





 





 



 } 




2723




 





 





 





 





 





 



   




2724




 





 





 





 





 





 



 # ============================ 




2725




 





 





 





 





 





 



 # reconstruct(schema, rows, top, cols, constraints, nesting, aliasstruct) 




2726




 





 





 





 





 





 



 # 




2727




 





 





 





 





 





 



 # mainly called by: selectall_stag(...) 




2728




 





 





 





 





 





 



 # 




2729




 





 





 





 





 





 



 # takes an array of rows (ie the result of an SQL query, probably 




2730




 





 





 





 





 





 



 # involving JOINs, which is a denormalised relation) and 




2731




 





 





 





 





 





 



 # decomposes this relation into a tree structure 




2732




 





 





 





 





 





 



 # 




2733




 





 





 





 





 





 



 # in order to do this, it requires schema information, and a nesting 




2734




 





 





 





 





 





 



 # through the implicit result graph to build a tree 




2735




 





 





 





 





 





 



 # ============================ 




2736




 





 





 





 





 





 



 sub reconstruct { 




2737




 





 





 





 





 





 



     my $self = shift; 




2738




 





 





 





 





 





 



     my $tree = Data::Stag->new(); 




2739




 





 





 





 





 





 



     my ($schema,         # OPTIONAL - meta data on relation 




2740




 





 





 





 





 





 



         $rows,           # REQUIRED - relation R - array-of-array 




2741




 





 





 





 





 





 



         $top,            # OPTIONAL - root node name 




2742




 





 





 





 





 





 



         $cols,           # REQUIRED - array of stag nodes per column of R 




2743




 





 





 





 





 





 



         $constraints,    # NOT USED!!! 




2744




 





 





 





 





 





 



         $nesting,           # REQUIRED - tree representing decomposed schema 




2745




 





 





 





 





 





 



         $aliasstruct,    # OPTIONAL - renaming of columns in R 




2746




 





 





 





 





 





 



         $aliaspolicy) = 




2747




 





 





 





 





 





 



           rearrange([qw(schema 




2748




 





 





 





 





 





 



                         rows 




2749




 





 





 





 





 





 



                         top 




2750




 





 





 





 





 





 



                         cols 




2751




 





 





 





 





 





 



                         constraints 




2752




 





 





 





 





 





 



                         nesting 




2753




 





 





 





 





 





 



                         alias 




2754




 





 





 





 





 





 



                         aliaspolicy)], @_); 




2755




 





 





 





 





 





 



   




2756




 





 





 





 





 





 



     $aliaspolicy = 'nest' unless $aliaspolicy; 




2757




 





 





 





 





 





 



   




2758




 





 





 





 





 





 



     # --- get the schema --- 




2759




 





 





 





 





 





 



     # 




2760




 





 





 





 





 





 



     # $schema is a stag representing the schema 




2761




 





 





 





 





 





 



     # of the input releation R (not the schema of 




2762




 





 





 





 





 





 



     # the db that produced it.... hmm, this could 




2763




 





 





 





 





 





 



     # be misleading) 




2764




 





 





 





 





 





 



     # 




2765




 





 





 





 





 





 



     # it conforms to the following stag-struct: 




2766




 





 





 





 





 





 



     # 




2767




 





 





 





 





 





 



     #'(schema 




2768




 





 





 





 





 





 



     #  (top? "RECORDSET-ELEMENT-NAME") 




2769




 





 





 





 





 





 



     #  (cols? 




2770




 





 





 





 





 





 



     #   (col+ 




2771




 





 





 





 





 





 



     #    (relation "RELATION-NAME") 




2772




 





 





 





 





 





 



     #    (name     "COLUMN-NAME") 




2773




 





 





 





 





 





 



     #    )) 




2774




 





 





 





 





 





 



     #  (nesting? 




2775




 





 





 





 





 





 



     #   (* "NESTING-TREE"))) 




2776




 





 





 





 





 





 



     # 




2777




 





 





 





 





 





 



     # each column represents the  




2778




 





 





 





 





 





 



      




2779




 





 





 





 





 





 



     if (!$schema) { 




2780




 





 





 





 





 





 



         $schema = $tree->new(schema=>[]); 




2781




 





 





 





 





 





 



     } 




2782




 





 





 





 





 





 



     if (!ref($schema)) { 




2783




 





 





 





 





 





 



         # it is a string - parse it 




2784




 





 





 





 





 





 



         # (assume sxpr) 




2785




 





 





 





 





 





 



         $schema = $tree->from('sxprstr', $schema); 




2786




 





 





 





 





 





 



     } 




2787




 





 





 





 





 





 



   




2788




 





 





 





 





 





 



     # TOP - this is the element name 




2789




 





 





 





 





 





 



     # to group the structs under. 




2790




 





 





 





 





 





 



     # [override if specified explicitly] 




2791




 





 





 





 





 





 



     if ($top) { 




2792




 





 





 





 





 





 



         stag_set($schema, 'top', $top); 




2793




 





 





 





 





 





 



     } 




2794




 





 





 





 





 





 



 #    $top = $schema->get_top || "set"; 




2795




 





 





 





 





 





 



     if (!$top) { 




2796




 





 





 





 





 





 



 	if ($nesting) { 




2797




 





 





 





 





 





 



 	    # use first element in nesting 




2798




 





 





 





 





 





 



 	    $top = $nesting->element; 




2799




 





 





 





 





 





 



 	} 




2800




 





 





 





 





 





 



 	else { 




2801




 





 





 





 





 





 



 	    $top = 'set'; 




2802




 





 





 





 





 





 



 	} 




2803




 





 





 





 





 





 



     } 




2804




 





 





 





 





 





 



     my $topstruct = $tree->new($top, []); 




2805




 





 





 





 





 





 



   




2806




 





 





 





 





 





 



     # COLS - this is the columns (attribute names) 




2807




 





 





 





 





 





 



     # in the order they appear 




2808




 





 





 





 





 





 



     # [override if specified explicitly] 




2809




 





 





 





 





 





 



     if ($cols) { 




2810




 





 





 





 





 





 



         my @ncols = 




2811




 





 





 





 





 





 



           map { 




2812




 





 





 





 





 





 



               if (ref($_)) { 




2813




 





 





 





 





 





 



                   $_ 




2814




 





 





 





 





 





 



               } 




2815




 





 





 





 





 





 



               else { 




2816




 





 





 





 





 





 



                   # presume it's a string 




2817




 





 





 





 





 





 



                   # format = RELATION.ATTRIBUTENAME 




2818




 





 





 





 





 





 



                   if (/(\w+)\.(\w+)/) { 




2819




 





 





 





 





 





 



                       $tree->new(col=>[ 




2820




 





 





 





 





 





 



                                        [relation=>$1], 




2821




 





 





 





 





 





 



                                        [name=>$2]]); 




2822




 





 





 





 





 





 



                   } 




2823




 





 





 





 





 





 



                   elsif (/(\w+)/) { 




2824




 





 





 





 





 





 



                       confess("Not implemented yet - must specify tbl for $_"); 




2825




 





 





 





 





 





 



                       $tree->new(col=>[ 




2826




 





 





 





 





 





 



                                        [relation=>'unknown'], 




2827




 





 





 





 





 





 



                                        [name=>$2]]); 




2828




 





 





 





 





 





 



                   } 




2829




 





 





 





 





 





 



                   else { 




2830




 





 





 





 





 





 



                       confess "I am confused by this column: $_"; 




2831




 





 





 





 





 





 



                   } 




2832




 





 





 





 





 





 



               } 




2833




 





 





 





 





 





 



           } @$cols; 




2834




 





 





 





 





 





 



         $schema->set_cols([@ncols]); 




2835




 





 





 





 





 





 



     } 




2836




 





 





 





 





 





 



   




2837




 





 





 





 





 





 



   




2838




 





 





 





 





 





 



     # NESTING - this is the tree structure in 




2839




 





 





 





 





 





 



     # which the relations are structured 




2840




 





 





 





 





 





 



     # [override if specified explicitly] 




2841




 





 





 





 





 





 



     if ($nesting) { 




2842




 





 





 





 





 





 



         if (ref($nesting)) { 




2843




 





 





 





 





 





 



         } 




2844




 





 





 





 





 





 



         else { 




2845




 





 





 





 





 





 



             $nesting = $tree->from('sxprstr', $nesting); 




2846




 





 





 





 





 





 



         } 




2847




 





 





 





 





 





 



         $schema->set_nesting([$nesting]); 




2848




 





 





 





 





 





 



     } 




2849




 





 





 





 





 





 



     else { 




2850




 





 





 





 





 





 



         $nesting = $schema->sgetnode_nesting; 




2851




 





 





 





 





 





 



     } 




2852




 





 





 





 





 





 



     if (!$nesting) { 




2853




 





 





 





 





 





 



         confess("no nesting!"); 




2854




 





 





 





 





 





 



     } 




2855




 





 





 





 





 





 



   




2856




 





 





 





 





 





 



     # --- alias structure --- 




2857




 





 





 





 





 





 



     # 




2858




 





 





 





 





 





 



     # use this to get a hash map of alias => baserelation  




2859




 





 





 





 





 





 



   




2860




 





 





 





 





 





 



     ($aliasstruct) = $schema->getnode_aliases unless $aliasstruct; 




2861




 





 





 





 





 





 



     if ($aliasstruct && !ref($aliasstruct)) { 




2862




 





 





 





 





 





 



         $aliasstruct = $tree->from('sxprstr', $aliasstruct); 




2863




 





 





 





 





 





 



     } 




2864




 





 





 





 





 





 



     my @aliases = (); 




2865




 





 





 





 





 





 



     if ($aliasstruct && $aliaspolicy !~ /^a/i) { 




2866




 





 





 





 





 





 



         @aliases = $aliasstruct->getnode_alias; 




2867




 





 





 





 





 





 



     } 




2868




 





 





 





 





 





 



     my %alias2baserelation = 




2869




 





 





 





 





 





 



       map { 




2870




 





 





 





 





 





 



           $_->sget_name => $_->sget_table 




2871




 





 





 





 





 





 



       } @aliases; 




2872




 





 





 





 





 





 



   




2873




 





 





 





 





 





 



     # column headings; (ie all columns in R) 




2874




 





 





 





 





 





 



     my @cols = $schema->sgetnode_cols->getnode_col(); 




2875




 





 





 





 





 





 



   




2876




 





 





 





 





 





 



     # --- primary key info --- 




2877




 





 





 





 





 





 



   




2878




 





 





 





 





 





 



     # set the primary key for each relation (one per relation); 




2879




 





 





 





 





 





 



     # the default is *all* the columns in that relation 




2880




 





 





 





 





 





 



     my %pkey_by_relationname = ();  # eg {person => [person_id] 




2881




 





 





 





 





 





 



     my %cols_by_relationname = ();  # eg {person => [person_id, fname, lname] 




2882




 





 





 





 





 





 



   




2883




 





 





 





 





 





 



     # loop through all columns in R, setting above hash maps 




2884




 





 





 





 





 





 



     foreach my $col (@cols) { 




2885




 





 





 





 





 





 



   




2886




 





 





 





 





 





 



 	# the stag struct for each $col looks like this: 




2887




 





 





 





 





 





 



 	# 




2888




 





 





 





 





 





 



 	#   (col+ 




2889




 





 





 





 





 





 



 	#    (relation "RELATION-NAME") 




2890




 





 





 





 





 





 



 	#    (name     "COLUMN-NAME") 




2891




 





 





 





 





 





 



 	#    )) 




2892




 





 





 





 





 





 



 	 




2893




 





 





 





 





 





 



         my $relationname = $col->get_relation; 




2894




 





 





 





 





 





 



         my $colname = $col->get_name; 




2895




 





 





 





 





 





 



   




2896




 





 





 





 





 





 



 	# pkey defaults to all columns in a relation 




2897




 





 





 





 





 





 



 	# (we may override this later) 




2898




 





 





 





 





 





 



         $pkey_by_relationname{$relationname} = [] 




2899




 





 





 





 





 





 



           unless $pkey_by_relationname{$relationname}; 




2900




 





 





 





 





 





 



         push(@{$pkey_by_relationname{$relationname}}, 




2901




 





 





 





 





 





 



              $colname); 




2902




 





 





 





 





 





 



   




2903




 





 





 





 





 





 



 	# all columns in a relation 




2904




 





 





 





 





 





 



 	# (note: same as default PK) 




2905




 





 





 





 





 





 



         $cols_by_relationname{$relationname} = [] 




2906




 





 





 





 





 





 



           unless $cols_by_relationname{$relationname}; 




2907




 





 





 





 





 





 



         push(@{$cols_by_relationname{$relationname}}, 




2908




 





 





 





 





 





 



              $colname); 




2909




 





 





 





 





 





 



     } 




2910




 





 





 





 





 





 



     my @relationnames = keys %pkey_by_relationname; 




2911




 





 





 





 





 





 



   




2912




 





 





 





 





 





 



     # override PK if explicitly set as a constraint 




2913




 





 





 





 





 





 



     my @pks = $schema->findnode("primarykey"); 




2914




 





 





 





 





 





 



     foreach my $pk (@pks) { 




2915




 





 





 





 





 





 



   




2916




 





 





 





 





 





 



 	# $pk looks like this: 




2917




 





 





 





 





 





 



 	# 




2918




 





 





 





 





 





 



 	# '(primarykey 




2919




 





 





 





 





 





 



 	#   (relation "R-NAME") 




2920




 





 





 





 





 





 



 	#   (col+ "COL-NAME")) 




2921




 





 





 





 





 





 



   




2922




 





 





 





 





 





 



         my $relationname = $pk->get_relation; 




2923




 





 





 





 





 





 



         my @cols = $pk->get_col; 




2924




 





 





 





 





 





 



   




2925




 





 





 





 





 





 



 	# the hash %pkey_by_relationname should 




2926




 





 





 





 





 





 



 	# be keyed by the named relations, not the 




2927




 





 





 





 





 





 



 	# base relations 




2928




 





 





 





 





 





 



         my @aliasnames = 




2929




 





 





 





 





 





 



           grep {  




2930




 





 





 





 





 





 



               $alias2baserelation{$_} eq $relationname  




2931




 





 





 





 





 





 



           } keys %alias2baserelation; 




2932




 





 





 





 





 





 



   




2933




 





 





 





 





 





 



         # relation is not aliased 




2934




 





 





 





 





 





 



         if (!@aliasnames) { 




2935




 





 





 





 





 





 



             @aliasnames = ($relationname); 




2936




 





 





 





 





 





 



         } 




2937




 





 





 





 





 





 



         foreach (@aliasnames) { 




2938




 





 





 





 





 





 



             $pkey_by_relationname{$_} = [@cols]; 




2939




 





 





 





 





 





 



         } 




2940




 





 





 





 





 





 



     } 




2941




 





 





 





 





 





 



   




2942




 





 





 





 





 





 



     # ------------------ 




2943




 





 





 





 





 





 



     # 




2944




 





 





 





 





 





 



     # loop through denormalised rows, 




2945




 





 





 





 





 





 



     # putting the columns into their 




2946




 





 





 





 





 





 



     # respecive relations 




2947




 





 





 





 





 





 



     # 




2948




 





 





 





 





 





 



     # eg 




2949




 





 





 





 





 





 



     # 




2950




 





 





 





 





 





 



     #  <----- a ----->   <-- b --> 




2951




 





 





 





 





 





 



     #  a.1   a.2   a.3   b.1   b.2 




2952




 





 





 





 





 





 



     # 




2953




 





 





 





 





 





 



     # algorithm: 




2954




 





 





 





 





 





 



     #  use nesting/tree to walk through 




2955




 





 





 





 





 





 



     # 




2956




 





 





 





 





 





 



     # ------------------ 




2957




 





 





 





 





 





 



   




2958




 





 





 





 





 





 



     #~~~  keep a hash of all relations by their primary key vals 




2959




 





 





 





 





 





 



     #~~~   outer key = relationname 




2960




 





 





 





 





 





 



     #~~~   inner key = pkval 




2961




 





 





 





 





 





 



     #~~~   hash val  = relation structure 




2962




 





 





 





 





 





 



     #~~~ my %all_relation_hh = (); 




2963




 





 





 





 





 





 



     #~~~ foreach my $relationname (@relationnames) { 




2964




 





 





 





 





 





 



     #~~~     $all_relation_hh{$relationname} = {}; 




2965




 





 





 





 





 





 



     #~~~ } 




2966




 





 





 





 





 





 



   




2967




 





 





 





 





 





 



     #~~~ keep an array of all relations 




2968




 





 





 





 





 





 



     #~~~  outer key = relationname 




2969




 





 





 





 





 





 



     #~~~  inner array = ordered list of relations 




2970




 





 





 





 





 





 



     #~~~    my %all_relation_ah = (); 




2971




 





 





 





 





 





 



     #~~~    foreach my $relationname (keys %pkey_by_relationname) { 




2972




 





 





 





 





 





 



     #~~~        $all_relation_ah{$relationname} = []; 




2973




 





 





 





 





 





 



     #~~~    } 




2974




 





 





 





 





 





 



   




2975




 





 





 





 





 





 



     # start at top of nesting tree 




2976




 





 





 





 





 





 



     # 




2977




 





 





 





 





 





 



     # a typical nesting tree may look like this: 




2978




 





 





 





 





 





 



     # 




2979




 





 





 





 





 





 



     # '(tableA 




2980




 





 





 





 





 





 



     #   (tableB "1") 




2981




 





 





 





 





 





 



     #   (tableC 




2982




 





 





 





 





 





 



     #    (tableD "1"))) 




2983




 





 





 





 





 





 



     # 




2984




 





 





 





 





 





 



     # terminals ie "1" are ignored 




2985




 





 





 





 





 





 



   




2986




 





 





 





 





 





 



     my ($first_in_nesting) = $nesting->subnodes; 




2987




 





 





 





 





 





 



     if (!$first_in_nesting) { 




2988




 





 





 





 





 





 



         $first_in_nesting = $nesting; 




2989




 





 





 





 





 





 



     } 




2990




 





 





 





 





 





 



     my $fipname = $first_in_nesting ? $first_in_nesting->name : ''; 




2991




 





 





 





 





 





 



   




2992




 





 





 





 





 





 



     # recursive hash representing tree 




2993




 





 





 





 





 





 



     # 




2994




 





 





 





 





 





 



     # $record = 




2995




 





 





 





 





 





 



     #   {child_h => {  




2996




 





 





 





 





 





 



     #                $relation_name* => { 




2997




 





 





 





 





 





 



     #                                   $pk_val => $record 




2998




 





 





 





 





 





 



     #                                  } 




2999




 





 





 





 





 





 



     #               }, 




3000




 





 





 





 





 





 



     #    struct => $stag_obj 




3001




 





 





 





 





 





 



     #   } 




3002




 





 





 





 





 





 



     # 




3003




 





 





 





 





 





 



     # this is recursively constructed using the make_a_tree() method 




3004




 





 





 





 





 





 



     # below. the nesting tree (see above) is traversed depth first, 




3005




 





 





 





 





 





 



     # constructing both the child_h hash and the resulting Stag 




3006




 





 





 





 





 





 



     # structure. 




3007




 





 





 





 





 





 



   




3008




 





 





 





 





 





 



     my $top_record_h =  




3009




 





 





 





 





 





 



       { 




3010




 





 





 





 





 





 



        child_h=>{ $fipname ? ($fipname=>{}) : () }, 




3011




 





 





 





 





 





 



        struct=>$topstruct 




3012




 





 





 





 





 





 



       }; 




3013




 





 





 





 





 





 



     # loop through rows in R 




3014




 





 





 





 





 





 



     foreach my $row (@$rows) { 




3015




 





 





 





 





 





 



         my @colvals = @$row; 




3016




 





 





 





 





 





 



   




3017




 





 





 





 





 





 



         # keep a record of all table names in 




3018




 





 





 





 





 





 



         # this row from R 




3019




 





 





 





 





 





 



         my %current_relation_h = (); 




3020




 





 





 





 





 





 



         for (my $i=0; $i<@cols; $i++) { 




3021




 





 





 





 





 





 



             my $colval = $colvals[$i]; 




3022




 





 





 





 





 





 



             my $col = $cols[$i]; 




3023




 





 





 





 





 





 



             my $relationname = $col->get_relation; 




3024




 





 





 





 





 





 



             my $colname = $col->get_name; 




3025




 





 





 





 





 





 



             my $relation = $current_relation_h{$relationname}; 




3026




 





 





 





 





 





 



             if (!$relation) { 




3027




 





 





 





 





 





 



                 $relation = {}; 




3028




 





 





 





 





 





 



                 $current_relation_h{$relationname} = $relation; 




3029




 





 





 





 





 





 



             } 




3030




 





 





 





 





 





 



             $relation->{$colname} = $colval; 




3031




 





 





 





 





 





 



         } 




3032




 





 





 





 





 





 



   




3033




 





 





 





 





 





 



 #	print "ROW=@$row\n"; 




3034




 





 





 





 





 





 



 #	dmp(\%pkey_by_relationname); 




3035




 





 





 





 





 





 



 #	dmp($top_record_h); 




3036




 





 





 





 





 





 



   




3037




 





 





 





 





 





 



         # we now have a hash of hashes - 




3038




 





 





 





 





 





 



         #  outer keyed by relation id 




3039




 





 





 





 





 





 



         #  inner keyed by relation attribute name 




3040




 





 





 





 





 





 



          




3041




 





 





 





 





 





 



         # traverse depth first down nesting; 




3042




 





 





 





 





 





 



         # add new nodes as children of the parent 




3043




 





 





 





 





 





 



         $self->make_a_tree($tree, 




3044




 





 





 





 





 





 



                            $top_record_h, 




3045




 





 





 





 





 





 



                            $first_in_nesting, 




3046




 





 





 





 





 





 



                            \%current_relation_h, 




3047




 





 





 





 





 





 



                            \%pkey_by_relationname, 




3048




 





 





 





 





 





 



                            \%cols_by_relationname, 




3049




 





 





 





 





 





 



                            \%alias2baserelation, 




3050




 





 





 





 





 





 



                            $aliaspolicy); 




3051




 





 





 





 





 





 



     } 




3052




 





 





 





 





 





 



     return $topstruct; 




3053




 





 





 





 





 





 



 } 




3054




 





 





 





 





 





 



 *norm = \&reconstruct; 




3055




 





 





 





 





 





 



 *normalise = \&reconstruct; 




3056




 





 





 





 





 





 



 *normalize = \&reconstruct; 




3057




 





 





 





 





 





 



   




3058




 





 





 





 





 





 



 # ============================ 




3059




 





 





 





 





 





 



 # make_a_tree(...) RECURSIVE 




3060




 





 





 





 





 





 



 # 




3061




 





 





 





 





 





 



 # called by: reconstruct(...) 




3062




 





 





 





 





 





 



 # 




3063




 





 





 





 





 





 



 # ============================ 




3064




 





 





 





 





 





 



 sub make_a_tree { 




3065




 





 





 





 





 





 



     my $self = shift; 




3066




 





 





 





 





 





 



     my $tree = shift; 




3067




 





 





 





 





 





 



     my $parent_rec_h = shift; 




3068




 





 





 





 





 





 



     my $node = shift; 




3069




 





 





 





 





 





 



     my %current_relation_h= %{shift ||{}}; 




3070




 





 





 





 





 





 



     my %pkey_by_relationname = %{shift ||{}}; 




3071




 





 





 





 





 





 



     my %cols_by_relationname = %{shift ||{}}; 




3072




 





 





 





 





 





 



     my %alias2baserelation = %{shift ||{}}; 




3073




 





 





 





 





 





 



     my $aliaspolicy = shift; 




3074




 





 





 





 





 





 



   




3075




 





 





 





 





 





 



     my $relationname = $node->name; 




3076




 





 





 





 





 





 



     my $relationrec = $current_relation_h{$relationname}; 




3077




 





 





 





 





 





 



     my $pkcols = $pkey_by_relationname{$relationname}; 




3078




 





 





 





 





 





 



     my $rec; # this is the next node down in the hash tree 




3079




 





 





 





 





 





 



   




3080




 





 





 





 





 





 



     if (!$pkcols || !@$pkcols) { 




3081




 





 





 





 





 





 



 	# if we have no columns for a particular part of 




3082




 





 





 





 





 





 



 	# the nesting through the relation, it means it 




3083




 





 





 





 





 





 



 	# was ommitted from the select clause - just skip 




3084




 





 





 





 





 





 



 	# this part of the nesting. 




3085




 





 





 





 





 





 



 	# 




3086




 





 





 





 





 





 



 	# for example: SELECT a.*, b.* FROM a NJ a_to_b NJ b 




3087




 





 





 





 





 





 



 	# the default nesting will be: [a [a_to_b [b]]] 




3088




 





 





 





 





 





 



 	# the relation R will have columns: 




3089




 





 





 





 





 





 



 	# a.c1    a.c2    b.c1   b.c2 




3090




 





 





 





 





 





 



 	# 




3091




 





 





 





 





 





 



 	# we want to build a resulting structure like this: 




3092




 





 





 





 





 





 



 	# (a 




3093




 





 





 





 





 





 



 	#  (c1 "x") (c2 "y") 




3094




 





 





 





 





 





 



 	#  (b 




3095




 





 





 





 





 





 



 	#   (c1 "a") (c2 "b"))) 




3096




 





 





 





 





 





 



 	# 




3097




 





 





 





 





 





 



 	# so we just miss out a_to_b in the nesting, because it 




3098




 





 





 





 





 





 



 	# has no columns in the relation R. 




3099




 





 





 





 





 





 



 	$rec = $parent_rec_h; 




3100




 





 





 





 





 





 



     } 




3101




 





 





 





 





 





 



     else { 




3102




 





 





 





 





 





 



   




3103




 





 





 





 





 





 



 	my $pkval =  




3104




 





 





 





 





 





 



 	  CORE::join("\t", 




3105




 





 





 





 





 





 



 		     map { 




3106




 





 





 





 





 





 



 			 esctab($relationrec->{$_} || '') 




3107




 





 





 





 





 





 



 		     } @$pkcols); 




3108




 





 





 





 





 





 



   




3109




 





 





 





 





 





 



 	$rec = $parent_rec_h->{child_h}->{$relationname}->{$pkval}; 




3110




 





 





 





 





 





 



   




3111




 





 





 





 





 





 



 	if (!$rec) { 




3112




 





 





 





 





 





 



 	    my $relationcols = $cols_by_relationname{$relationname}; 




3113




 





 





 





 





 





 



 	    my $has_non_null_val = grep {defined($relationrec->{$_})} @$relationcols; 




3114




 





 





 





 





 





 



 	    return unless $has_non_null_val; 




3115




 





 





 





 





 





 



 	    my $relationstruct = 




3116




 





 





 





 





 





 



 	      $tree->new($relationname=>[ 




3117




 





 





 





 





 





 



 					 map { 




3118




 





 





 





 





 





 



 					     defined($relationrec->{$_}) ? [$_ => $relationrec->{$_}] : () 




3119




 





 





 





 





 





 



 					 } @$relationcols 




3120




 





 





 





 





 





 



 					]); 




3121




 





 





 





 





 





 



 	    my $parent_relationstruct = $parent_rec_h->{struct}; 




3122




 





 





 





 





 





 



 	    if (!$parent_relationstruct) { 




3123




 





 





 





 





 





 



 		confess("no parent for $relationname"); 




3124




 





 





 





 





 





 



 	    } 




3125




 





 





 





 





 





 



          




3126




 





 





 





 





 





 



 	    # if we have an aliased relation, add an extra 




3127




 





 





 





 





 





 



 	    # level of nesting 




3128




 





 





 





 





 





 



 	    my $baserelation = $alias2baserelation{$relationname}; 




3129




 





 





 





 





 





 



 	    if ($baserelation) { 




3130




 





 





 





 





 





 



   




3131




 





 





 





 





 





 



                 # $aliaspolicy eq 'nest' or 't' 




3132




 





 





 





 





 





 



                 # nest base relations inside an alias node 




3133




 





 





 





 





 





 



                 # OR use table name in place of alias name 




3134




 





 





 





 





 





 



                 if ($aliaspolicy =~ /^t/i) { 




3135




 





 





 





 





 





 



                     stag_add($parent_relationstruct, 




3136




 





 





 





 





 





 



                              $baserelation, 




3137




 





 





 





 





 





 



                              $relationstruct->data); 




3138




 





 





 





 





 





 



                 } 




3139




 





 





 





 





 





 



                 else { 




3140




 





 





 





 





 





 



                     # nest 




3141




 





 





 





 





 





 



                     my $baserelationstruct = 




3142




 





 





 





 





 





 



                       Data::Stag->new($baserelation => 




3143




 





 





 





 





 





 



                                       $relationstruct->data); 




3144




 





 





 





 





 





 



                     stag_add($parent_relationstruct, 




3145




 





 





 





 





 





 



                              $relationname, 




3146




 





 





 





 





 





 



                              [$baserelationstruct]); 




3147




 





 





 





 





 





 



                 } 




3148




 





 





 





 





 





 



 	    } else { 




3149




 





 





 





 





 





 



                 # either no aliases, or $aliaspolicy eq 'a' 




3150




 





 





 





 





 





 



                 # (in which case columns already mapped to aliases) 




3151




 





 





 





 





 





 



                 stag_add($parent_relationstruct, 




3152




 





 





 





 





 





 



                          $relationstruct->name, 




3153




 





 





 





 





 





 



                          $relationstruct->data); 




3154




 





 





 





 





 





 



 	    } 




3155




 





 





 





 





 





 



 	    $rec = 




3156




 





 





 





 





 





 



 	      {struct=>$relationstruct, 




3157




 





 





 





 





 





 



 	       child_h=>{}}; 




3158




 





 





 





 





 





 



 	    foreach ($node->subnodes) { 




3159




 





 





 





 





 





 



 		# keep index of children by PK 




3160




 





 





 





 





 





 



 		$rec->{child_h}->{$_->name} = {}; 




3161




 





 





 





 





 





 



 	    } 




3162




 





 





 





 





 





 



 	    $parent_rec_h->{child_h}->{$relationname}->{$pkval} = $rec; 




3163




 





 





 





 





 





 



 	} 




3164




 





 





 





 





 





 



     } 




3165




 





 





 





 





 





 



     foreach ($node->subnodes) { 




3166




 





 





 





 





 





 



         $self->make_a_tree($tree, 




3167




 





 





 





 





 





 



 			   $rec, 




3168




 





 





 





 





 





 



 			   $_, 




3169




 





 





 





 





 





 



 			   \%current_relation_h, 




3170




 





 





 





 





 





 



 			   \%pkey_by_relationname, 




3171




 





 





 





 





 





 



 			   \%cols_by_relationname, 




3172




 





 





 





 





 





 



 			   \%alias2baserelation, 




3173




 





 





 





 





 





 



                            $aliaspolicy); 




3174




 





 





 





 





 





 



     } 




3175




 





 





 





 





 





 



 } 




3176




 





 





 





 





 





 



   




3177




 





 





 





 





 





 



   




3178




 





 





 





 





 





 



 # -------- GENERAL SUBS ----------- 




3179




 





 





 





 





 





 



   




3180




 





 





 





 





 





 



 sub esctab { 




3181




 





 





 





 





 





 



     my $w=shift; 




3182




 





 





 





 





 





 



     $w =~ s/\t/__MAGICTAB__/g; 




3183




 





 





 





 





 





 



     $w; 




3184




 





 





 





 





 





 



 } 




3185




 





 





 





 





 





 



   




3186




 





 





 





 





 





 



 sub makesql { 




3187




 





 





 





 





 





 



     my $self = shift; 




3188




 





 





 





 





 





 



     my ($table, 




3189




 





 





 





 





 





 



         $where, 




3190




 





 





 





 





 





 



         $select, 




3191




 





 





 





 





 





 



         $order, 




3192




 





 





 





 





 





 



         $group, 




3193




 





 





 





 





 





 



         $distinct) = 




3194




 





 





 





 





 





 



           rearrange([qw(table 




3195




 





 





 





 





 





 



                         where 




3196




 





 





 





 





 





 



                         select 




3197




 





 





 





 





 





 



                         order 




3198




 





 





 





 





 





 



                         group 




3199




 





 





 





 





 





 



                         distinct)], @_); 




3200




 





 





 





 





 





 



   




3201




 





 





 





 





 





 



     confess("must specify table") unless $table; 




3202




 





 





 





 





 





 



   




3203




 





 





 





 





 





 



     # array of tables 




3204




 





 





 





 





 





 



     if (ref($table)) { 




3205




 





 





 





 





 





 



         if (ref($table) eq "HASH") { 




3206




 





 





 





 





 





 



             $table = 




3207




 





 





 





 





 





 



               [ 




3208




 





 





 





 





 





 



                map { 




3209




 





 





 





 





 





 



                    "$table->{$_} AS $_" 




3210




 





 





 





 





 





 



                } keys %$table 




3211




 





 





 





 





 





 



               ]; 




3212




 





 





 





 





 





 



         } 




3213




 





 





 





 





 





 



     } 




3214




 





 





 





 





 





 



     else { 




3215




 





 





 





 





 





 



         $table = [$table]; 




3216




 





 





 





 





 





 



     } 




3217




 





 





 





 





 





 



   




3218




 





 





 





 





 





 



     $where = [] unless $where; 




3219




 





 





 





 





 





 



     # array of ANDed where clauses 




3220




 





 





 





 





 





 



     if (ref($where)) { 




3221




 





 





 





 





 





 



         if (ref($where) eq "HASH") { 




3222




 





 





 





 





 





 



             $where = 




3223




 





 





 





 





 





 



               [ 




3224




 





 





 





 





 





 



                map { 




3225




 





 





 





 





 





 



                    "$_ = ".$self->quote($where->{$_}) 




3226




 





 





 





 





 





 



                } keys %$where 




3227




 





 





 





 





 





 



               ]; 




3228




 





 





 





 





 





 



         } 




3229




 





 





 





 





 





 



     } 




3230




 





 





 





 





 





 



     else { 




3231




 





 





 





 





 





 



         $where = [$where]; 




3232




 





 





 





 





 





 



     } 




3233




 





 





 





 





 





 



   




3234




 





 





 





 





 





 



     $select = ['*'] unless $select; 




3235




 





 





 





 





 





 



     # array of SELECT cols 




3236




 





 





 





 





 





 



     if (ref($select)) { 




3237




 





 





 





 





 





 



         if (ref($select) eq "HASH") { 




3238




 





 





 





 





 





 



             $select = 




3239




 





 





 





 





 





 



               [ 




3240




 





 





 





 





 





 



                map { 




3241




 





 





 





 





 





 



                    "$select->{$_} AS $_" 




3242




 





 





 





 





 





 



                } keys %$select 




3243




 





 





 





 





 





 



               ]; 




3244




 





 





 





 





 





 



         } 




3245




 





 





 





 





 





 



     } 




3246




 





 





 





 





 





 



     else { 




3247




 





 





 





 





 





 



         $select = [$select]; 




3248




 





 





 





 





 





 



     } 




3249




 





 





 





 





 





 



   




3250




 





 





 





 





 





 



     $order = [] unless $order; 




3251




 





 





 





 





 





 



     # array of order tables 




3252




 





 





 





 





 





 



     if (ref($order)) { 




3253




 





 





 





 





 





 



         if (ref($order) eq "HASH") { 




3254




 





 





 





 





 





 



             confess("order must be an array"); 




3255




 





 





 





 





 





 



         } 




3256




 





 





 





 





 





 



     } 




3257




 





 





 





 





 





 



     else { 




3258




 





 





 





 





 





 



         $order = [$order]; 




3259




 





 





 





 





 





 



     } 




3260




 





 





 





 





 





 



   




3261




 





 





 





 





 





 



     $group = [] unless $group; 




3262




 





 





 





 





 





 



     # array of group tables 




3263




 





 





 





 





 





 



     if (ref($group)) { 




3264




 





 





 





 





 





 



         if (ref($group) eq "HASH") { 




3265




 





 





 





 





 





 



             confess("group must be an array"); 




3266




 





 





 





 





 





 



         } 




3267




 





 





 





 





 





 



     } 




3268




 





 





 





 





 





 



     else { 




3269




 





 





 





 





 





 



         $group = [$group]; 




3270




 





 





 





 





 





 



     } 




3271




 





 





 





 





 





 



   




3272




 





 





 





 





 





 



     $distinct = $distinct ? '' : ' DISTINCT'; 




3273




 





 





 





 





 





 



     my $sql = 




3274




 





 





 





 





 





 



       sprintf("SELECT%s %s FROM %s%s%s", 




3275




 





 





 





 





 





 



               $distinct, 




3276




 





 





 





 





 





 



               join(', ', @$select), 




3277




 





 





 





 





 





 



               join(', ', @$table), 




3278




 





 





 





 





 





 



               (scalar(@$where) ?  




3279




 





 





 





 





 





 



                ' WHERE '.join(' AND ', @$where) : ''), 




3280




 





 





 





 





 





 



               (scalar(@$group) ?  




3281




 





 





 





 





 





 



                ' GROUP BY '.join(', ', @$group) : ''), 




3282




 





 





 





 





 





 



               (scalar(@$order) ?  




3283




 





 





 





 





 





 



                ' ORDER BY '.join(', ', @$order) : ''), 




3284




 





 





 





 





 





 



              ); 




3285




 





 





 





 





 





 



     return $sql; 




3286




 





 





 





 





 





 



 } 




3287




 





 





 





 





 





 



   




3288




 





 





 





 





 





 



   




3289




 





 





 





 





 





 



   




3290




 





 





 





 





 





 



 sub selectval { 




3291




 





 





 





 





 





 



     my $self = shift; 




3292




 





 





 





 





 





 



     trace(0, "@_") if $TRACE; 




3293




 





 





 





 





 





 



     return $self->dbh->selectcol_arrayref(@_)->[0]; 




3294




 





 





 





 





 





 



 } 




3295




 





 





 





 





 





 



   




3296




 





 





 





 





 





 



 sub insertrow { 




3297




 





 





 





 





 





 



     my $self = shift; 




3298




 





 





 





 





 





 



     my ($table, $colvalh, $pkcol) = @_; 




3299




 





 





 





 





 





 



        




3300




 





 





 





 





 





 



     my $driver = $self->dbh->{Driver}->{Name}; 




3301




 





 





 





 





 





 



     my @cols = keys %$colvalh; 




3302




 





 





 





 





 





 



     my @vals =  




3303




 





 





 





 





 





 



       map { 




3304




 





 





 





 





 





 



           defined($_) ? $colvalh->{$_} : undef 




3305




 





 





 





 





 





 



       } @cols; 




3306




 





 





 





 





 





 



     my @placeholders = map { '?' } @vals; 




3307




 





 





 





 





 





 



     my $sql = 




3308




 





 





 





 





 





 



       sprintf("INSERT INTO %s (%s) VALUES (%s)", 




3309




 





 





 





 





 





 



               $table, 




3310




 





 





 





 





 





 



               join(", ", @cols), 




3311




 





 





 





 





 





 



               #join(", ", @vals), 




3312




 





 





 





 





 





 



               join(", ", @placeholders), 




3313




 





 





 





 





 





 



              ); 




3314




 





 





 





 





 





 



     if (!@cols) { 




3315




 





 





 





 





 





 



 	$sql = "INSERT INTO $table DEFAULT VALUES"; 




3316




 





 





 





 





 





 



     } 




3317




 





 





 





 





 





 



   




3318




 





 





 





 





 





 



     trace(0, "SQL:$sql") if $TRACE; 




3319




 





 





 





 





 





 



     my $succeeded = 0; 




3320




 





 





 





 





 





 



     eval { 




3321




 





 





 





 





 





 



         my $sth = $self->dbh->prepare($sql); 




3322




 





 





 





 





 





 



         my $rval = $sth->execute(@vals); 




3323




 





 





 





 





 





 



         $succeeded = 1 if defined $rval; 




3324




 





 





 





 





 





 



     }; 




3325




 





 





 





 





 





 



     if ($@) { 




3326




 





 





 





 





 





 



 	if ($self->force) { 




3327




 





 





 





 





 





 



 	    # what about transactions?? 




3328




 





 





 





 





 





 



 	    $self->warn("IN SQL: $sql\nWARNING: $@\n"); 




3329




 





 





 





 





 





 



             return; 




3330




 





 





 





 





 





 



 	} 




3331




 





 





 





 





 





 



 	else { 




3332




 





 





 





 





 





 



 	    confess $@; 




3333




 





 





 





 





 





 



 	} 




3334




 





 





 





 





 





 



     } 




3335




 





 





 





 





 





 



     return unless $succeeded; 




3336




 





 





 





 





 





 



     my $pkval; 




3337




 





 





 





 





 





 



     if ($pkcol) { 




3338




 





 





 





 





 





 



         # primary key value may have been specified in the xml 




3339




 





 





 





 





 





 



         # (this is necessary for non-surrogate pks in tables that 




3340




 





 





 





 





 





 



         #  are to be linked to via foreign keys) 




3341




 





 





 





 





 





 



         $pkval = $colvalh->{$pkcol}; 




3342




 





 





 





 





 





 



   




3343




 





 





 





 





 





 



         # pk was not supplied - perhaps this is a SERIAL/AUTO_INCREMENT 




3344




 





 





 





 





 





 



         # column (ie surrogate integer primary key) 




3345




 





 





 





 





 





 



         if (!$pkval) { 




3346




 





 





 





 





 





 



             # assume pk is a SERIAL / AUTO_INCREMENT 




3347




 





 





 





 





 





 



             if ($driver eq 'Pg') { 




3348




 





 





 





 





 





 



                 my $seqn = sprintf("%s_%s_seq", 




3349




 





 





 





 





 





 



                                    $table, 




3350




 





 





 





 





 





 



                                    $pkcol); 




3351




 





 





 





 





 





 



                 $pkval  = $self->selectval("select currval('$seqn')");         




3352




 





 





 





 





 





 



                 trace(0, "CURRVAL $seqn = $pkval     [Pg]") if $TRACE; 




3353




 





 





 





 





 





 



             } 




3354




 





 





 





 





 





 



             # this doesn't work on older 




3355




 





 





 





 





 





 



             # versions of DBI/DBD::mysql 




3356




 





 





 





 





 





 



             # seems to have been fixed Oct 2004 release 




3357




 





 





 





 





 





 



             elsif ($driver eq 'mysql') { 




3358




 





 





 





 





 





 



                 $pkval = $self->dbh->last_insert_id(undef,undef,$table,$pkcol); 




3359




 





 





 





 





 





 



                 trace(0, "CURRVAL mysql_insert_id $pkval   [mysql]") if $TRACE; 




3360




 





 





 





 





 





 



             } 




3361




 





 





 





 





 





 



             else { 




3362




 





 





 





 





 





 



                 $pkval  = $self->selectval("select max($pkcol) from $table"); 




3363




 





 





 





 





 





 



             } 




3364




 





 





 





 





 





 



         } 




3365




 





 





 





 





 





 



         trace(0, "PKVAL = $pkval") if $TRACE; 




3366




 





 





 





 





 





 



     } 




3367




 





 





 





 





 





 



     return $pkval; 




3368




 





 





 





 





 





 



 } 




3369




 





 





 





 





 





 



   




3370




 





 





 





 





 





 



 sub updaterow { 




3371




 





 





 





 





 





 



     my $self = shift; 




3372




 





 





 





 





 





 



     my ($table, $set, $where) = @_; 




3373




 





 





 





 





 





 



   




3374




 





 





 





 





 





 



     confess("must specify table") unless $table; 




3375




 





 





 





 





 





 



   




3376




 





 





 





 





 





 



     my $dbh = $self->dbh; 




3377




 





 





 





 





 





 



   




3378




 





 





 





 





 





 



     # array of WHERE cols 




3379




 





 





 





 





 





 



     if (ref($where)) { 




3380




 





 





 





 





 





 



         if (ref($where) eq "HASH") { 




3381




 





 





 





 





 





 



             $where = 




3382




 





 





 





 





 





 



               [ 




3383




 





 





 





 





 





 



                map { 




3384




 





 





 





 





 





 



                    "$_ = ".$dbh->quote($where->{$_}) 




3385




 





 





 





 





 





 



                } keys %$where 




3386




 





 





 





 





 





 



               ]; 




3387




 





 





 





 





 





 



         } 




3388




 





 





 





 





 





 



     } 




3389




 





 





 





 





 





 



     else { 




3390




 





 





 





 





 





 



         $where = [$where]; 




3391




 





 





 





 





 





 



     } 




3392




 





 





 





 





 





 



     confess("must specify constraints") unless @$where; 




3393




 





 





 





 





 





 



   




3394




 





 





 





 





 





 



     confess("must set update vals") unless $set; 




3395




 





 





 





 





 





 



     my @bind = (); 




3396




 





 





 





 





 





 



     # array of SET colvals 




3397




 





 





 





 





 





 



     if (ref($set)) { 




3398




 





 





 





 





 





 



         if (ref($set) eq "HASH") { 




3399




 





 





 





 





 





 



             $set = 




3400




 





 





 





 





 





 



               [ 




3401




 





 





 





 





 





 



                map { 




3402




 





 





 





 





 





 



                    push(@bind, defined $set->{$_} ? $set->{$_} : 'NULL'); 




3403




 





 





 





 





 





 



                    "$_ = ?" 




3404




 





 





 





 





 





 



                } keys %$set 




3405




 





 





 





 





 





 



               ]; 




3406




 





 





 





 





 





 



         } 




3407




 





 





 





 





 





 



     } 




3408




 





 





 





 





 





 



     else { 




3409




 





 





 





 





 





 



         $set = [$set]; 




3410




 





 





 





 





 





 



     } 




3411




 





 





 





 





 





 



      




3412




 





 





 





 





 





 



     my $sql = 




3413




 





 





 





 





 





 



       sprintf("UPDATE %s SET %s WHERE %s", 




3414




 





 





 





 





 





 



               $table, 




3415




 





 





 





 





 





 



               join(', ', @$set), 




3416




 





 





 





 





 





 



               join(' AND ', @$where), 




3417




 





 





 





 





 





 



              ); 




3418




 





 





 





 





 





 



     trace(0, "SQL:$sql [",join(', ',@bind)."]") if $TRACE; 




3419




 





 





 





 





 





 



   




3420




 





 





 





 





 





 



     my $sth = $dbh->prepare($sql) || confess($sql."\n\t".$dbh->errstr); 




3421




 





 





 





 





 





 



     return $sth->execute(@bind) || confess($sql."\n\t".$sth->errstr); 




3422




 





 





 





 





 





 



 } 




3423




 





 





 





 





 





 



   




3424




 





 





 





 





 





 



 sub deleterow { 




3425




 





 





 





 





 





 



     my $self = shift; 




3426




 





 





 





 





 





 



     my ($table, $where) = @_; 




3427




 





 





 





 





 





 



   




3428




 





 





 





 





 





 



     confess("must specify table") unless $table; 




3429




 





 





 





 





 





 



   




3430




 





 





 





 





 





 



     my $dbh = $self->dbh; 




3431




 





 





 





 





 





 



   




3432




 





 





 





 





 





 



     # array of WHERE cols 




3433




 





 





 





 





 





 



     if (ref($where)) { 




3434




 





 





 





 





 





 



         if (ref($where) eq "HASH") { 




3435




 





 





 





 





 





 



             $where = 




3436




 





 





 





 





 





 



               [ 




3437




 





 





 





 





 





 



                map { 




3438




 





 





 





 





 





 



                    "$_ = ".$dbh->quote($where->{$_}) 




3439




 





 





 





 





 





 



                } keys %$where 




3440




 





 





 





 





 





 



               ]; 




3441




 





 





 





 





 





 



         } 




3442




 





 





 





 





 





 



     } 




3443




 





 





 





 





 





 



     else { 




3444




 





 





 





 





 





 



         $where = [$where]; 




3445




 





 





 





 





 





 



     } 




3446




 





 





 





 





 





 



     confess("must specify constraints") unless @$where; 




3447




 





 





 





 





 





 



   




3448




 





 





 





 





 





 



     my $sql = 




3449




 





 





 





 





 





 



       sprintf("DELETE FROM %s WHERE %s", 




3450




 





 





 





 





 





 



               $table, 




3451




 





 





 





 





 





 



               join(' AND ', @$where), 




3452




 





 





 





 





 





 



              ); 




3453




 





 





 





 





 





 



     trace(0, "SQL:$sql") if $TRACE; 




3454




 





 





 





 





 





 



   




3455




 





 





 





 





 





 



     my $sth = $dbh->prepare($sql) || confess($sql."\n\t".$dbh->errstr); 




3456




 





 





 





 





 





 



     return $sth->execute() || confess($sql."\n\t".$sth->errstr); 




3457




 





 





 





 





 





 



 } 




3458




 





 





 





 





 





 



   




3459




 





 





 





 





 





 



 #$::RD_HINT = 1; 




3460




 





 





 





 





 





 



   




3461




 





 





 





 





 





 



 $::RD_AUTOACTION = q { [@item] }; 




3462




 





 





 





 





 





 



 sub selectgrammar { 




3463




 





 





 





 





 





 



     return q[ 




3464




 





 





 





 





 





 



   




3465




 





 





 





 





 





 



              { 




3466




 





 





 





 





 





 



               use Data::Dumper; 




3467




 





 





 





 





 





 



               use Data::Stag; 




3468




 





 





 





 





 





 



               sub N { 




3469




 





 





 





 





 





 



                   Data::Stag->new(@_); 




3470




 





 





 





 





 





 



               } 




3471




 





 





 





 





 





 



           } 




3472




 





 





 





 





 





 



          ] 




3473




 





 





 





 





 





 



        . 




3474




 





 





 





 





 





 



          q[ 




3475




 





 





 





 





 





 



   




3476




 





 





 





 





 





 



          selectstmts: selectstmt ';' selectstmts 




3477




 





 





 





 





 





 



          selectstmts: selectstmt 




3478




 





 





 





 





 





 



            #           selectstmt: /select/i selectcols /from/i fromtables 




3479




 





 





 





 





 





 



          selectstmt: /select/i selectq(?) selectcols /from/i fromtables where(?) group(?) having(?) combiner(?) order(?) limit(?) offset(?)  




3480




 





 





 





 





 





 



            { 




3481




 





 





 





 





 





 



                N(select => [ 




3482




 





 





 





 





 





 



                             [qual => $item{'selectq'}[0]], 




3483




 





 





 





 





 





 



                             [cols => $item[3]], 




3484




 





 





 





 





 





 



                             [from => $item[5]], 




3485




 





 





 





 





 





 



 #                            [where => $item[6]], 




3486




 





 





 





 





 





 



 #                            [group => $item{'group'}[0]], 




3487




 





 





 





 





 





 



 #                            [having => $item{'having'}[0]], 




3488




 





 





 





 





 





 



                            ]); 




3489




 





 





 





 





 





 



            } 




3490




 





 





 





 





 





 



            |  




3491




 





 





 





 





 





 



          selectq: /all/i | /distinct/i  




3492




 





 





 





 





 





 



                { $item[1] } 




3493




 





 





 





 





 





 



            |                 




3494




 





 





 





 





 





 



 #	 as: /\s+as\s+/i 




3495




 





 





 





 





 





 



 	 as: /as/i 




3496




 





 





 





 





 





 



          selectcols: selectexpr /\,/ selectcols 




3497




 





 





 





 





 





 



            { [$item[1], @{$item[3]}] } 




3498




 





 





 





 





 





 



            |  




3499




 





 





 





 





 





 



          selectcols: selectexpr 




3500




 





 





 





 





 





 



              { [$item[1]] } 




3501




 





 





 





 





 





 



            |  




3502




 





 





 





 





 





 



          selectexpr: bselectexpr as aliasname 




3503




 





 





 





 





 





 



            { 




3504




 





 





 





 





 





 



                my $col = $item{bselectexpr}; 




3505




 





 





 





 





 





 



                $col->set_alias($item{aliasname}->[1]); 




3506




 





 





 





 





 





 



                $col; 




3507




 





 





 





 





 





 



            } 




3508




 





 





 





 





 





 



            |  




3509




 





 





 





 





 





 



 	 selectexpr: bselectexpr 




3510




 





 





 





 





 





 



             { $item[1] } 




3511




 





 





 





 





 





 



             |  




3512




 





 





 





 





 





 



          bselectexpr: funccall 




3513




 





 





 





 





 





 



            { $item[1] } 




3514




 





 





 





 





 





 



            |  




3515




 





 





 





 





 





 



          bselectexpr: selectcol 




3516




 





 





 





 





 





 



            { $item[1] } 




3517




 





 





 





 





 





 



            |  




3518




 





 





 





 





 





 



   




3519




 





 





 





 





 





 



 	 selectcol: brackselectcol operator selectcol 




3520




 





 





 





 





 





 



 	   { 




3521




 





 





 





 





 





 



 	       N(col=>[ 




3522




 





 





 





 





 





 



 		       [func => [ 




3523




 





 





 





 





 





 



 				 [name => $item[2]->[1]], 




3524




 





 





 





 





 





 



 				 [args => [$item[1],$item[3]]] 




3525




 





 





 





 





 





 



 				] 




3526




 





 





 





 





 





 



 		       ] 




3527




 





 





 





 





 





 



 		      ]); 




3528




 





 





 





 





 





 



 	   } 




3529




 





 





 





 





 





 



 ###	   { $item[1]} 




3530




 





 





 





 





 





 



 	   |  




3531




 





 





 





 





 





 



 	 selectcol: brackselectcol 




3532




 





 





 





 





 





 



 	   { $item[1]} 




3533




 





 





 





 





 





 



 	   |  




3534




 





 





 





 





 





 



   




3535




 





 





 





 





 





 



 	 brackselectcol: '(' selectcol ')'  




3536




 





 





 





 





 





 



 	   { $item[2]} 




3537




 





 





 





 





 





 



 	   |  




3538




 





 





 





 





 





 



   




3539




 





 





 





 





 





 



 	 brackselectcol: bselectcol 




3540




 





 





 





 





 





 



 	   { $item[1]} 




3541




 





 





 





 





 





 



 	   |  




3542




 





 





 





 





 





 



   




3543




 





 





 





 





 





 



          bselectcol: /(\w+)\.(\w+)/ 




3544




 





 





 





 





 





 



            { N(col=>[ 




3545




 





 





 





 





 





 



                      [name => $item[1]], 




3546




 





 





 





 





 





 



 		     [table=>$1], 




3547




 





 





 





 





 





 



                     ])  




3548




 





 





 





 





 





 



            } 




3549




 





 





 





 





 





 



            |  




3550




 





 





 





 





 





 



          bselectcol: /(\w+)\.\*/ 




3551




 





 





 





 





 





 



            { N(col=>[ 




3552




 





 





 





 





 





 



                      [name => $item[1]], 




3553




 





 





 





 





 





 



                      [table=>$1], 




3554




 





 





 





 





 





 



                     ])  




3555




 





 





 





 





 





 



            } 




3556




 





 





 





 





 





 



            |  




3557




 





 





 





 





 





 



          bselectcol: /\*/ 




3558




 





 





 





 





 





 



            { N(col=>[ 




3559




 





 





 





 





 





 



                      [name => $item[1]] 




3560




 





 





 





 





 





 



                     ])  




3561




 





 





 





 





 





 



            } 




3562




 





 





 





 





 





 



            |  




3563




 





 





 





 





 





 



          bselectcol: /\w+/ 




3564




 





 





 





 





 





 



            { N(col=>[ 




3565




 





 





 





 





 





 



                      [name => $item[1]] 




3566




 





 





 





 





 





 



                     ])  




3567




 





 





 





 





 





 



            } 




3568




 





 





 





 





 





 



            |  




3569




 





 





 





 





 





 



          bselectcol: expr 




3570




 





 





 





 





 





 



            { N(col=>[ 




3571




 





 





 





 





 





 



                      [expr => $item[1]] 




3572




 





 





 





 





 





 



                     ]) } 




3573




 





 





 





 





 





 



            |  




3574




 





 





 





 





 





 



          funccall: funcname '(' distinct(?) selectcols ')'  




3575




 





 





 





 





 





 



            { 




3576




 





 





 





 





 





 



             my $col = N(col=>[ 




3577




 





 





 





 





 





 



                               [func => [ 




3578




 





 





 





 





 





 



                                         [name => $item[1]->[1]], 




3579




 





 





 





 





 





 



                                         [args => $item[4]] 




3580




 





 





 





 





 





 



                                        ] 




3581




 





 





 





 





 





 



                               ] 




3582




 





 





 





 





 





 



                              ]); 




3583




 





 





 





 





 





 



             $col; 




3584




 





 





 





 





 





 



            } 




3585




 





 





 





 





 





 



            |  




3586




 





 





 





 





 





 



   




3587




 





 





 





 





 





 



          distinct: /distinct/i 




3588




 





 





 





 





 





 



   




3589




 





 





 





 





 





 



 	 operator: '+' | '-' | '*' | '/' | '||' 




3590




 





 





 





 





 





 



 	    




3591




 





 





 





 





 





 



   




3592




 





 





 





 





 





 



          fromtables: jtable 




3593




 





 





 





 





 





 



            { [$item[1]] } 




3594




 





 





 





 





 





 



            |  




3595




 





 





 





 





 





 



          jtable: join_jtable 




3596




 





 





 





 





 





 



            { $item[1] } 




3597




 





 





 





 





 





 



            |  




3598




 





 





 





 





 





 



          join_jtable: qual_jtable jointype join_jtable  




3599




 





 





 





 





 





 



            {  




3600




 





 





 





 





 





 



                shift @{$item[2]}; 




3601




 





 





 





 





 





 



                my $j = 




3602




 





 





 





 





 





 



                  N(composite=>[ 




3603




 





 





 





 





 





 



                                [ctype=>"@{$item[2]}"], 




3604




 





 





 





 





 





 



                                [first=>[$item[1]]], 




3605




 





 





 





 





 





 



                                [second=>[$item[3]]] 




3606




 





 





 





 





 





 



                               ]); 




3607




 





 





 





 





 





 



                $j; 




3608




 





 





 





 





 





 



          } 




3609




 





 





 





 





 





 



            |  




3610




 





 





 





 





 





 



          join_jtable: qual_jtable 




3611




 





 





 





 





 





 



            { $item[1] } 




3612




 





 





 





 





 





 



            |  




3613




 





 





 





 





 





 



          qual_jtable: alias_jtable joinqual  




3614




 





 





 





 





 





 



            {  




3615




 





 





 





 





 





 



                my $j = $item[1]; 




3616




 





 





 





 





 





 



                $j->setnode_qual($item[2]); 




3617




 





 





 





 





 





 



                $j; 




3618




 





 





 





 





 





 



            } 




3619




 





 





 





 





 





 



            |  




3620




 





 





 





 





 





 



          qual_jtable: alias_jtable 




3621




 





 





 





 





 





 



            { $item[1] } 




3622




 





 





 





 





 





 



            |  




3623




 





 





 





 





 





 



          alias_jtable: brack_jtable /as\s+/i aliasname  




3624




 





 





 





 





 





 



            {  




3625




 





 





 





 





 





 



                my $j = $item[1]; 




3626




 





 





 





 





 





 



                $j->set_alias($item[3][1]); 




3627




 





 





 





 





 





 



                $j; 




3628




 





 





 





 





 





 



            } 




3629




 





 





 





 





 





 



            |  




3630




 





 





 





 





 





 



          alias_jtable: brack_jtable 




3631




 





 





 





 





 





 



            { $item[1] } 




3632




 





 





 





 





 





 



            |  




3633




 





 





 





 





 





 



          brack_jtable: '(' jtable ')'  




3634




 





 





 





 





 





 



            { $item[2] } 




3635




 





 





 





 





 





 



            |  




3636




 





 





 





 





 





 



          brack_jtable: table 




3637




 





 





 





 





 





 



            { N(leaf=>[[name=>$item[1]->[1]]]) } 




3638




 





 





 





 





 





 



            |  




3639




 





 





 





 





 





 



   




3640




 





 





 





 





 





 



          joinqual: /on\s+/i bool_expr 




3641




 





 





 





 





 





 



            { N(qual => [ 




3642




 





 





 





 





 





 



                         [type=>'on'], 




3643




 





 





 





 





 





 



                         [expr=>"@{$item[2]}"] 




3644




 





 





 





 





 





 



                        ]) 




3645




 





 





 





 





 





 



          } 




3646




 





 





 





 





 





 



            |  




3647




 





 





 





 





 





 



          joinqual: /using\s+/i '(' cols ')' 




3648




 





 





 





 





 





 



            { N(qual =>[ 




3649




 





 





 





 





 





 



                        [type=>'using'], 




3650




 





 





 





 





 





 



                        [expr=>"@{$item[3]}"] 




3651




 





 





 





 





 





 



                       ]) 




3652




 





 





 





 





 





 



          } 




3653




 





 





 





 





 





 



            |  




3654




 





 





 





 





 





 



   




3655




 





 





 





 





 





 



          table: tablename 




3656




 





 





 





 





 





 



            { $item[1] } 




3657




 





 





 





 





 





 



            |  




3658




 





 





 





 





 





 



   




3659




 





 





 





 





 





 



          funcname: /\w+/ 




3660




 





 





 





 





 





 



          tablename: /\w+/ 




3661




 





 





 





 





 





 



          aliasname: /\w+/ 




3662




 





 





 





 





 





 



   




3663




 





 





 





 





 





 



   




3664




 





 





 





 





 





 



          cols: col(s) 




3665




 





 





 





 





 





 



          col: /\w+\.\w+/ 




3666




 





 





 





 





 





 



          col: /\w+/ 




3667




 





 





 





 





 





 



   




3668




 





 





 





 





 





 



          jointype: /\,/ 




3669




 





 





 





 





 





 



          jointype: /natural/i bjointype /join/i 




3670




 





 





 





 





 





 



          jointype: /natural/i /join/i 




3671




 





 





 





 





 





 



          jointype: bjointype /join/i 




3672




 





 





 





 





 





 



          jointype: /join/i 




3673




 





 





 





 





 





 



          bjointype: /inner/i 




3674




 





 





 





 





 





 



          bjointype: lrf(?) /outer/i 




3675




 





 





 





 





 





 



          lrf: /left/i | /right/i | /full/i 




3676




 





 





 





 





 





 



          bjointype: /cross/i 




3677




 





 





 





 





 





 



   




3678




 





 





 





 





 





 



          number: float | int 




3679




 





 





 





 





 





 



          float: /\d*\.?\d+/ 'e' sign int 




3680




 





 





 





 





 





 



          float: /\d*\.\d+/ 




3681




 





 





 





 





 





 



          int: /\d+/ 




3682




 





 





 





 





 





 



          string: /\'.*?\'/ 




3683




 





 





 





 





 





 



          sign: '+' | '-' 




3684




 





 





 





 





 





 



             




3685




 





 





 





 





 





 



          exprs: '(' exprs ')' 




3686




 





 





 





 





 





 



          exprs: expr ',' exprs 




3687




 





 





 





 





 





 



          exprs: expr 




3688




 





 





 





 





 





 



   




3689




 





 





 





 





 





 



 	   # bool_expr - eg in where clause 




3690




 





 





 





 





 





 



          bool_expr: not_bool_expr boolop bool_expr | not_bool_expr 




3691




 





 





 





 





 





 



          not_bool_expr: '!' brack_bool_expr | brack_bool_expr 




3692




 





 





 





 





 





 



          brack_bool_expr: '(' bool_expr ')' | bool_exprprim 




3693




 





 





 





 





 





 



          bool_exprprim: boolval | expr 




3694




 





 





 





 





 





 



          boolval: /true/i | /false/i | /null/i 




3695




 





 





 





 





 





 



   




3696




 





 





 





 





 





 



          expr: brack_expr op expr | brack_expr 




3697




 





 





 





 





 





 



          brack_expr: '(' expr ')' | exprprim 




3698




 





 





 





 





 





 



          exprprim: col | val 




3699




 





 





 





 





 





 



          val: number | string 




3700




 





 





 





 





 





 



             




3701




 





 





 





 





 





 



          op: /not\s+/i /like\s+/i 




3702




 





 





 





 





 





 



          op: /like\s+/i 




3703




 





 





 





 





 





 



          op: /is\s+/i /not\s+/i 




3704




 





 





 





 





 





 



          op: /is\s+/i 




3705




 





 





 





 





 





 



          op: '=' | '!=' | '<>' | '<=' | '>=' | '<' | '>' 




3706




 





 





 





 





 





 



          boolop: /and\s+/i | /or\s+/i | /not\s+/i 




3707




 





 





 





 





 





 



   




3708




 





 





 





 





 





 



 #           where: /where/i /.*/ 




3709




 





 





 





 





 





 



            where: /where/i bool_expr 




3710




 





 





 





 





 





 



            group: /group/i /by/i exprs 




3711




 





 





 





 





 





 



            having: /having/i /.*/ 




3712




 





 





 





 





 





 



            combiner: combinekwd selectstmt 




3713




 





 





 





 





 





 



            combinekwd: /union/i | /intersect/i | /update/i 




3714




 





 





 





 





 





 



            order: /order/i /by/i orderexprs 




3715




 





 





 





 





 





 



            orderexprs: orderexpr ',' orderexprs 




3716




 





 





 





 





 





 



            orderexprs: orderexpr 




3717




 





 





 





 





 





 



            orderexpr: expr /asc/i 




3718




 





 





 





 





 





 



            orderexpr: expr /desc/i 




3719




 





 





 





 





 





 



            orderexpr: expr /using/i op 




3720




 





 





 





 





 





 



            orderexpr: expr 




3721




 





 





 





 





 





 



            limit: /limit/i /\w+/ 




3722




 





 





 





 





 





 



            offset: /offset/i /\d+/ 




3723




 





 





 





 





 





 



             ]; 




3724




 





 





 





 





 





 



 } 




3725




 





 





 





 





 





 



   




3726




 





 





 





 





 





 



 no strict 'refs'; 




3727




 





 





 





 





 





 



 sub AUTOLOAD { 




3728




 





 





 





 





 





 



     my $self = shift; 




3729




 





 





 





 





 





 



     my @args = @_; 




3730




 





 





 





 





 





 



   




3731




 





 





 





 





 





 



     my $name = $AUTOLOAD; 




3732




 





 





 





 





 





 



     $name =~ s/.*://;   # strip fully-qualified portion 




3733




 





 





 





 





 





 



   




3734




 





 





 





 





 





 



     if ($name eq "DESTROY") { 




3735




 





 





 





 





 





 



 	# we dont want to propagate this!! 




3736




 





 





 





 





 





 



 	return; 




3737




 





 





 





 





 





 



     } 




3738




 





 





 





 





 





 



      




3739




 





 





 





 





 





 



     unless ($self->isa("DBIx::DBStag")) { 




3740




 





 





 





 





 





 



         confess("no such subroutine $name"); 




3741




 





 





 





 





 





 



     } 




3742




 





 





 





 





 





 



     if ($self->dbh) { 




3743




 





 





 





 





 





 



 	if ($TRACE) {  




3744




 





 





 





 





 





 



 	    # the following check may impair performance 




3745




 





 





 





 





 





 



 	    if (grep { ref($_) } @args) { 




3746




 





 





 





 





 





 



 		$self->throw("cannot quote @args"); 




3747




 





 





 





 





 





 



 	    } 




3748




 





 





 





 





 





 



 	} 




3749




 





 





 





 





 





 



 	if ($self->dbh->can($name)) { 




3750




 





 





 





 





 





 



 	    return $self->dbh->$name(@args); 




3751




 





 





 





 





 





 



 	} 




3752




 





 





 





 





 





 



     } 




3753




 





 





 





 





 





 



     confess("no such method:$name)"); 




3754




 





 





 





 





 





 



 } 




3755




 





 





 





 





 





 



   




3756




 





 





 





 





 





 



 sub rearrange { 




3757




 





 





 





 





 





 



   my($order,@param) = @_; 




3758




 





 





 





 





 





 



   




3759




 





 





 





 





 





 



   # If there are no parameters, we simply wish to return 




3760




 





 





 





 





 





 



   # an undef array which is the size of the @{$order} array. 




3761




 





 





 





 





 





 



   return (undef) x $#{$order} unless @param; 




3762




 





 





 





 





 





 



   




3763




 





 





 





 





 





 



   # If we've got parameters, we need to check to see whether 




3764




 





 





 





 





 





 



   # they are named or simply listed. If they are listed, we 




3765




 





 





 





 





 





 



   # can just return them. 




3766




 





 





 





 





 





 



   return @param unless (defined($param[0]) && $param[0]=~/^-/); 




3767




 





 





 





 





 





 



   




3768




 





 





 





 





 





 



   # Now we've got to do some work on the named parameters. 




3769




 





 





 





 





 





 



   # The next few lines strip out the '-' characters which 




3770




 





 





 





 





 





 



   # preceed the keys, and capitalizes them. 




3771




 





 





 





 





 





 



   my $i; 




3772




 





 





 





 





 





 



   for ($i=0;$i<@param;$i+=2) { 




3773




 





 





 





 





 





 



       if (!defined($param[$i])) { 




3774




 





 





 





 





 





 



 	  cluck("Hmmm in $i ".CORE::join(";", @param)." == ".CORE::join(";",@$order)."\n"); 




3775




 





 





 





 





 





 



       } 




3776




 





 





 





 





 





 



       else { 




3777




 





 





 





 





 





 



 	  $param[$i]=~s/^\-//; 




3778




 





 





 





 





 





 



 	  $param[$i]=~tr/a-z/A-Z/; 




3779




 





 





 





 





 





 



       } 




3780




 





 





 





 





 





 



   } 




3781




 





 





 





 





 





 



    




3782




 





 





 





 





 





 



   # Now we'll convert the @params variable into an associative array. 




3783




 





 





 





 





 





 



   my(%param) = @param; 




3784




 





 





 





 





 





 



   




3785




 





 





 





 





 





 



   my(@return_array); 




3786




 





 





 





 





 





 



    




3787




 





 





 





 





 





 



   # What we intend to do is loop through the @{$order} variable, 




3788




 





 





 





 





 





 



   # and for each value, we use that as a key into our associative 




3789




 





 





 





 





 





 



   # array, pushing the value at that key onto our return array. 




3790




 





 





 





 





 





 



   my($key); 




3791




 





 





 





 





 





 



   




3792




 





 





 





 





 





 



   foreach $key (@{$order}) { 




3793




 





 





 





 





 





 



       $key=~tr/a-z/A-Z/; 




3794




 





 





 





 





 





 



       my($value) = $param{$key}; 




3795




 





 





 





 





 





 



       delete $param{$key}; 




3796




 





 





 





 





 





 



       push(@return_array,$value); 




3797




 





 





 





 





 





 



   } 




3798




 





 





 





 





 





 



    




3799




 





 





 





 





 





 



   # catch user misspellings resulting in unrecognized names 




3800




 





 





 





 





 





 



   my(@restkeys) = keys %param; 




3801




 





 





 





 





 





 



   if (scalar(@restkeys) > 0) { 




3802




 





 





 





 





 





 



        confess("@restkeys not processed in rearrange(), did you use a 




3803




 





 





 





 





 





 



        non-recognized parameter name ? "); 




3804




 





 





 





 





 





 



   } 




3805




 





 





 





 





 





 



   return @return_array; 




3806




 





 





 





 





 





 



 } 




3807




 





 





 





 





 





 



   




3808




 





 





 





 





 





 



 #sub loadschema { 




3809




 





 





 





 





 





 



 #    my $self = shift; 




3810




 





 





 





 





 





 



 #    my ($ddl, $ddlf, $dialect) =  




3811




 





 





 





 





 





 



 #      rearrange([qw(ddl ddlf dialect)], @_); 




3812




 





 





 





 





 





 



 #    if ($ddlf) { 




3813




 





 





 





 





 





 



 #        my $fh = FileHandle->new($ddlf) || $self->throw("no file $ddlf"); 




3814




 





 





 





 





 





 



 #        $ddl = join('',<$fh>); 




3815




 





 





 





 





 





 



 #        $fh->close; 




3816




 





 





 





 





 





 



 #    } 




3817




 





 





 





 





 





 



 #    $self->throw("no DDL") unless $ddl; 




3818




 





 





 





 





 





 



 #    if ($dialect) { 




3819




 





 





 





 





 





 



 #        my $driver = $self->{_driver} || 'Pg'; 




3820




 





 





 





 





 





 



 #        if ($driver ne $dialect) { 




3821




 





 





 





 





 





 



              




3822




 





 





 





 





 





 



 #        } 




3823




 





 





 





 





 





 



 #    } 




3824




 





 





 





 





 





 



 #} 




3825




 





 





 





 





 





 



   




3826




 





 





 





 





 





 



 1; 




3827




 





 





 





 





 





 



   




3828




 





 





 





 





 





 



 __END__
	    if ($fk eq 'id') {
		$fk = $element . '_id';
	    }

            # --SET SUBNODE FK--
            # it is necessarily true that each delayed-store subnode
            # must have some fk relationship back to the existing one
            # the subnode has a fk relation up to this one;
            # by default we assume that the subnode fk column is named
            # the same as the current pk. However, we check that this
            # is the case. If not, we deduce what the correct fk col is
            my $subtable =
              $dbschema->table($snname);
            if ($subtable->column($fk)) {
                # a fk col with the name as the current node pk col exists;
                # use it

                # do nothing - current value of $fk is fine
            }
            else {
                # deduce actual fk column
                # there should only be ONE subnode fk column UNSET;
                # this implicitly refers to the current node
                my @subcolumns = $subtable->columns;
                my @potential_fks = ();
                foreach my $subcolumn (@subcolumns) {
                    if ($self->is_fk_col($subcolumn) &&
                        !$self->is_pk_col($subcolumn)) {

                        # Definite foreign key
                        if (defined $sn->sget($subcolumn)) {
                            # already set
                        }
                        else {
                            push(@potential_fks, $subcolumn);
                        }
                    }
                }
                trace(0, "POTENTIAL FKS: @potential_fks");
                if (!@potential_fks) {
                    $self->throw("I do not know what to do with the current ".
                                 "pl val ($id). There does not appear to be ".
                                 "a $fk column in $snname, and all fks in ".
                                 "the subtable $snname are currently set");
                }
                if (@potential_fks > 1) {
                    $self->throw("There appear to be multiple potential fks ".
                                 "[ @potential_fks ]. I do not know which ".
                                 "to choose to assign the current pk val $id".
                                 " to");
                }
                $fk = shift @potential_fks;
            }
            # -- $fk value is set
            $sn->set($fk, $id);
            # -- $fk table assigned

            trace(0, "NOW TIME TO STORE [curr pk val = $id] [fkcol = $fk] ", $sn->xml) if $TRACE;
            # store subnode, passing in info on current node
            $self->_storenode($sn,{parent_pk_id=>$id,
                                   parent_element=>$element,
                                   assigned_node_h=>{$fk=>1}});
        }
    } # -- end of @delayed_store

    if ($current_attribute_node) {
        if ($id) {
            my $macro_id = $current_attribute_node->sget_id;
            if ($macro_id) {
                $self->macro_id_h->{$macro_id} = $id;
                trace(0, "SETTING MACRO ID MAP: $macro_id => $id") if $TRACE;
            }
            else {
            }
        }
    }

    return $id;
}

# --SQL directives embedded in XML--

sub _execute_sqlnode {
    my $self = shift;
    my $sqlnode = shift;
    if ($sqlnode->element eq '_sql') {
        my $dbh = $self->dbh;
        my $op = $sqlnode->get('@/op');
        my $col = $sqlnode->get('@/col');
        my $table = $sqlnode->get('@/from');
        my $match = $sqlnode->get('@/match');
        my @subnodes = grep {$_->element ne '@'} $sqlnode->kids;
        if ($op eq 'delete') {
            my $pkey = $sqlnode->get('@/pkey');
            trace(0,"deleting from $table");
            my @vals = map {$self->_execute_sqlnode($_)} @subnodes;
            # do iteratively rather than in 1 SQL stmt
            if (@vals) {
                my $sql =
                  sprintf("SELECT $pkey FROM $table WHERE $match IN (%s)",
                          join(", ",@vals));
                trace(0, "SQL: $sql");
                my $ids_to_delete =
                  $dbh->selectcol_arrayref($sql); # quote
                foreach my $id (@$ids_to_delete) {
                    my $delete_sql =
                      "DELETE FROM $table WHERE $pkey=$id";
                    trace(0,"SQL: $delete_sql");
                    $dbh->do($delete_sql);
                }
            }
        }
        elsif ($op eq "select") {
            my @vals = $sqlnode->get('.');
            my $sql =
              sprintf("SELECT $col FROM $table WHERE $match IN (%s)",
                      join(", ",map {$dbh->quote($_)} @vals));
            trace(0, "SQL: $sql");
            my $ids =
              $dbh->selectcol_arrayref($sql);
            trace(0,"id list in select: @$ids");
            return(@$ids);
        }
        else {
            $self->throw("Do not understand SQL directive: $op")
        }
    }
    else {
        return $sqlnode->data;
    }
    return;
}

sub _process_sql {
    my $self = shift;
    my $node = shift;
    my $element = $node->element;
    if ($element eq 'in') {

    }
    else {
        $self->throw("Do not understand SQL directive: $element")
    }
}

# -- QUERYING --

sub rmake_nesting {
    my $node = shift;

    if ($node->element eq 'composite') {
	my $first = $node->getnode_first;
	my $second = $node->getnode_second;
	my $head = rmake_nesting($first->data->[0]);
	my $tail = rmake_nesting($second->data->[0]);
	if ($head->isterminal) {
	    return
	      Data::Stag->new($head->element => [$tail]);
	}
	$head->addkid($tail);
	return $head;
    }
    elsif ($node->element eq 'leaf') {
	my $alias = $node->get_alias;
	my $tn = $alias || $node->get_name;
	return Data::Stag->new($tn=>1);
    }
    else {
	die;
    }
}

# if true, a metadata tag will be added to stag nodes selected from db
sub include_metadata {
    my $self = shift;
    $self->{_include_metadata} = shift if @_;
    return $self->{_include_metadata};
}


# last SQL SELECT statement executed
sub last_stmt {
    my $self = shift;
    $self->{_last_stmt} = shift if @_;
    return $self->{_last_stmt};
}

sub last_sql_and_args {
    my $self = shift;
    $self->{_last_sql_and_args} = shift if @_;
    return $self->{_last_sql_and_args};
}


sub sax_handler {
    my $self = shift;
    $self->{_sax_handler} = shift if @_;
    return $self->{_sax_handler};
}


# delegates to selectall_stag and turns tree to XML
sub selectall_xml {
    my $self = shift;
    my $stag = $self->selectall_stag(@_);
    return $stag->xml;
}

# delegates to selectall_stag and turns tree to SAX
# (candidate for optimisation - TODO - use event firing model)
sub selectall_sax {
    my $self = shift;
    my ($sql, $nesting, $h) =
      rearrange([qw(sql nesting handler)], @_);
    my $stag = $self->selectall_stag(@_);
    $h = $h || $self->sax_handler;
    if (!$h) {
	$self->throw("You must specify the sax handler;\n".
		     "Either use \$dbh->sax_handler(\$h), or \n".
		     "\$dbh->selectall_sax(-sql=>\$sql, handler->\$h)");
    }
    return $stag->sax($h);
}

# delegates to selectall_stag and turns tree to S-Expression
sub selectall_sxpr {
    my $self = shift;
    my $stag = $self->selectall_stag(@_);
    return $stag->sxpr;
}

# does not bother decomposing and nesting the results; just
# returns the denormalised table from the SQL query.
# arrayref of arrayrefs - rows x cols
# first row of rows is column headings
sub selectall_rows {
    my $self = shift;
    my ($sql, $nesting, $bind, $template) =
      rearrange([qw(sql nesting bind template)], @_);
    my $rows =
      $self->selectall_stag(-sql=>$sql,
			    -nesting=>$nesting,
			    -bind=>$bind,
			    -template=>$template,
			    -return_arrayref=>1,
			   );
    return $rows;
}

# ---------------------------------------
# selectall_stag(sql, nesting)
#
# Takes an sql string containing a SELECT statement,
# parses it to get the tree structure; this can be
# overridden with the nesting optional argument.
#
# The SELECT statement is executed, and the relations are
# transformed into a stag tree
#
# ---------------------------------------
sub selectall_stag {
    my $self = shift;
    my ($sql, $nesting, $bind, $template, $return_arrayref, $include_metadata, $aliaspolicy) =
      rearrange([qw(sql nesting bind template return_arrayref include_metadata aliaspolicy)], @_);
    my $prep_h = $self->prepare_stag(@_);
    my $cols = $prep_h->{cols};
    my $sth = $prep_h->{sth};
    my $exec_args = $prep_h->{exec_args};

    if (!defined($include_metadata)) {
        $include_metadata = $self->include_metadata;
    }

    # TODO - make this event based so we don't have to
    # load all into memory
    my $rows =
      $self->dbh->selectall_arrayref($sth, undef, @$exec_args);
    if ($return_arrayref) {
	my @hdrs = ();
	for (my $i=0; $i<@$cols; $i++) {
	    my $h = $prep_h->{col_aliases_ordered}->[$i] || $cols->[$i];
	    push(@hdrs, $h);
	}
	return [\@hdrs, @$rows];
    }

    trace(0, sprintf("Got %d rows\n", scalar(@$rows))) if $TRACE;
    # --- reconstruct tree from relations
    my $stag =
      $self->reconstruct(
                         -rows=>$rows,
                         -cols=>$cols,
                         -alias=>$prep_h->{alias},
                         -nesting=>$prep_h->{nesting},
                         -aliaspolicy=>$aliaspolicy,
                        );
    if ($include_metadata) {
        my ($last_sql, @sql_args) = @{$self->last_sql_and_args || []};
        my @kids = $stag->kids;
        my @bind_nodes;
        if ($bind && ref($bind) eq 'HASH') {
            @bind_nodes = (stag_unflatten(argset=>[%$bind]));
        }
        unshift(@kids,
                [dbstag_metadata=>[
                                   [sql=>$last_sql],
                                   [nesting=>$nesting],
                                   [template=>$template],
                                   @bind_nodes,
                                   (map {[exec_arg=>$_]} @sql_args)
                                  ]]);
        $stag->kids(@kids);
    }
    return $stag;
}

sub prepare_stag {
    my $self = shift;
    my ($sql, $nesting, $bind, $template, $return_arrayref, $aliaspolicy) =
      rearrange([qw(sql nesting bind template return_arrayref aliaspolicy)], @_);

    my $parser = $self->parser;

    my $sth;
    my @exec_args = ();
    if (ref($sql)) {
	$template = $sql;
    }
    if ($template) {
	if (!ref($template)) {
	    $template = $self->find_template($template);
	}
	($sql, @exec_args) = $template->get_sql_and_args($bind);
    }
    trace 0, "parsing_sql: $sql\n";

    # PRE-parse SQL statement for stag-specific extensions
    if ($sql =~ /(.*)\s+use\s+nesting\s*(.*)/si) {
	my ($pre, $post) = ($1, $2);
	my ($extracted, $remainder) =
	  extract_bracketed($post, '()');
        if ($nesting) {
            $self->throw("nestings clash: $nesting vs $extracted");
        }
	$nesting = Data::Stag->parsestr($extracted);
	$sql = "$pre $remainder";
    }


    # get the parsed SQL SELECT statement as a stag node
    my $stmt = $parser->selectstmt($sql);
    if (!$stmt) {
	# there was some error parsing the SQL;
	# DBI can probably give a better explanation.
	eval {
	    my $sth = $self->dbh->prepare($sql);

	};
	if ($@) {
	    $self->throw("SQL ERROR:\n$@");
	}
	# DBI accepted it - must be a bug in the DBStag grammar
	$self->throw("I'm sorry but the SQL statement you gave does\n".
		     "not conform to the more limited subset of SQL\n".
		     "that DBStag supports. Please see the DBStag docs\n".
		     "for details.\n".
		     "\n".
		     "Remember to check you explicitly declare all aliases\n".
		     "using AS\n\n\nSQL:$sql");
    }


    trace 0, "parsed_sql: $sql\n";
#    trace 0, $stmt->xml;
    my $dbschema = $self->dbschema;

    $self->last_stmt($stmt);

    # stag node of FROM part of SQL
    my $fromstruct = $stmt->get_from;

    # --- aliases ---

    # keep a hash of table aliases
    # KEY: table alias
    # VAL: base table
    # for example, 'SELECT * FROM person AS p'
    # will result in $alias_h = { p => person }
    my $alias_h = {};

    # build alias hash using FROM node
    foreach my $sn ($fromstruct->subnodes) {
        get_table_alias_map($sn, $alias_h);
    }

    # as well as an alias hash map,
    # keep an array of stag nodes representing all the aliases
    my @aliases = ();
    foreach my $alias (keys %$alias_h) {
        push(@aliases,
             Data::Stag->new(alias=>[
                                     [name=>$alias],
                                     [table=>$alias_h->{$alias}->[0]]
                                    ]));
    }
    my $aliasstruct = Data::Stag->new(alias=>[@aliases]);

    # --- nestings ---
    #
    # the cartesian product that results from a SELECT can
    # be turned into a tree - there is more than one tree to
    # choose from; eg with "x NJ y NJ z" we can have trees:
    # [x [y [z]]]
    # [x [y z]]
    # [z [x y]]
    # etc
    #
    # the actual allowed nestings through the graph is constrained
    # by the FK relationships; we do not utilise this yet (TODO!)
    # later the user need only specify the root. for now they
    # must specify the full nesting OR allow the bracket structure
    # of the joins...

    # if the user did not explicitly supply a nesting,
    # guess one from the bracket structure of the FROM
    # clause (see rmake_nesting)
    # [TODO: be more clever in guessing the nesting using FKs]
    if (!$nesting) {
	$nesting = Data::Stag->new(top=>1);
#	my $cons = rmake_cons($fromstruct->data->[0], $nesting);
	$nesting = rmake_nesting($fromstruct->data->[0]);
        $nesting = Data::Stag->new(top=>[$nesting]);
	trace(0, "\n\nNesting:\n%s\n\n",  $nesting->xml) if $TRACE;
    }
    if ($nesting && !ref($nesting)) {
        $nesting = Data::Stag->parsestr($nesting);
    }

    # keep an array of named relations used in the query -
    # the named relation is the alias if present;
    # eg
    # SELECT * FROM person AS p NATURAL JOIN department
    # the named relations here are 'p' and 'department'
    my @namedrelations = ();
    $fromstruct->iterate(sub {
			     my $n = shift;
			     if ($n->element eq 'leaf') {
                                 my $v = $n->sget_alias || $n->sget_name;
				 push(@namedrelations, $v)
			     }
			 });

    # --- fetch columns ---
    #
    # loop through all the columns in the SELECT clause
    # making them all of a standard form; eg dealing
    # with functions and '*' wildcards appropriately

    my @col_aliases_ordered = ();
    my @cols =
      map {
	  # $_ iterator variable is over the columns
	  # specified in the SELECT part of the query;
	  # each column is represented as a stag node

	  # column name
          my $name = $_->get_name;

	  # column alias, if exists
	  # eg in 'SELECT name AS n' the alias is 'n'
          my $col_alias = $_->get_alias;
	  push(@col_aliases_ordered, $col_alias);

	  # make the name the alias; prepend named relation if supplied.
	  # eg in 'SELECT person.name AS n' the name will become
	  # 'person.n'
	  if ($col_alias) {
	      $name = $col_alias;
	      if ($_->get_table) {
		  $name = $_->get_table . '.'. $name;
	      }
	  }

	  my $func = $_->getnode('func');

	  # from here on determines returned value of the
	  # map iteration:

	  if ($func) {
	      # a typical column node for a function looks like
	      # this:
	      #
	      #    (col
	      #      (func
	      #        (name "somefunc")
	      #        (args
	      #          (col
	      #            (name "x.foo")
	      #            (table "x")))))
	      #      (alias "myname"))
	      #
	      # if a function is included, and the function
	      # return value is aliased, use that alias;
	      # otherwise ...

	      my $funcname = $func->get_name;
	      # query the function stag node for the element
	      # 'col'
	      my ($col) =
		$func->where('col',
			     sub {shift->get_table});
	      my $table = $col_alias || $funcname;
	      if (!$col_alias) {
		  $col_alias = $funcname;
	      }
	      if ($col) {
		  $table = $col->get_table;
	      }
#	      if ($col_alias =~ /(\w+)__(\w+)/) {
#		  $table = $1;
#		  $col_alias = $2;
#	      }
	      $name = $table . '.' . $col_alias;
	      # return:
	      $name;
	  }
          elsif ($name =~ /^(\w+)\.\*$/) {
	      # if the column name is of the form
	      # RELATION.*, then replace the * with
	      # all the actual columns from the base relation
	      # RELATION
	      #
	      # the final result will be TABLE.col1, TABLE.col2,...

              my $tn = $1;
              my $tn_alias = $tn;

	      # use base relation name to introspect schema
              if ($alias_h->{$tn}) {
                  $tn = $alias_h->{$tn}->[0];
              }
              my $tbl = $dbschema->table(lc($tn));
              if (!$tbl) {
                  confess("No such table as $tn");
              }
	      # introspect schema to get columns for this table
              my @cns = $tbl->columns;

#	      trace(0, Dumper $tbl) if $TRACE;
	      trace(0, "TN:$tn ALIAS:$tn_alias COLS:@cns") if $TRACE;

	      # return:
              map { "$tn_alias.$_" } @cns;
          }
          elsif ($name =~ /^\*$/) {
	      # if the column name is '*' (ie select all)
	      # then replace the * with
	      # all the actual columns from the base relations in
	      # the query (use FROM clause)
	      #

              my %got = ();
              my @allcols =
                map {
                    my $tn = $_;
                    my $baserelname =
                      $alias_h->{$tn} ?
                        $alias_h->{$tn}->[0] : $tn;
                    my $tbl = $dbschema->table(lc($baserelname));
                    if (!$tbl) {
                        confess("Don't know anything about table:$tn\n".
                                "Maybe DBIx::DBSchema does not work for your DBMS?\n".
                                "If $tn is a view, you may need to modify DBIxLLDBSchema");
                    }
                    my @cns = $tbl->columns;
                    #                  @cns = grep { !$got{$_}++ } @cns;
                    map { "$tn.$_"} @cns;
                } @namedrelations;

              # This is a bit hacky; if the user specifies
              # SELECT * FROM... then there is no way
              # to introspect the actual column returned
              # using DBI->selectall_arrayref
              #
              # maybe we should selectall_hashref
              # instead? this is generally slower; also
              # even if we get it with a hashref, the
              # result can be ambiguous since DBI only
              # gives us the colun names back
              #
              # to get round this we just replace the *
              # in the user's query (ie in the actual SQL)
	      # with the full column list
              my $replace = join(', ', @allcols);
	      # rewrite SQL statement; assum only one instance of
	      # string '*' in these cases
              $sql =~ s/\*/$replace/;
	      # return:
              @allcols;
	  }
          else {
	      # no * wildcard in column, and not a function;
	      # just give back the node

	      # return:
              $name
          }
      } $stmt->sgetnode_cols->getnode_col;

    @cols =
      map {
	  if (/(\w+)__(\w+)/) {
	      "$1.$2";
	  }
	  else {
	      $_
	  }
      } @cols;

    # ---- end of column fetching ---

    trace(0, "COLS:@cols") if $TRACE;



    # --- execute SQL SELECT statement ---
    if ($template) {
	$sth = $template->cached_sth->{$sql};
	if (!$sth) {
	    $sth = $self->dbh->prepare($sql);
	    $template->cached_sth->{$sql} = $sth;
	}
#	($sql, $sth, @exec_args) =
#	  $template->prepare($self->dbh, $bind);
    }
    else {
        $sth = $self->dbh->prepare($sql);
    }
    my $sql_or_sth = $sql;
    if ($sth) {
	$sql_or_sth = $sth;
    }
    trace(0, "SQL:$sql") if $TRACE;
    trace(0, "Exec_args: @exec_args") if $TRACE && @exec_args;
    $self->last_sql_and_args([$sql, @exec_args]);
    return
      {
       sth=>$sth,
       exec_args=>\@exec_args,
       cols=>\@cols,
       col_aliases_ordered=>\@col_aliases_ordered,
       alias=>$aliasstruct,
       nesting=>$nesting
      };
}


# ============================
# get_table_alias_map(tablenode, alias hash)
#
# checks a tablenode (eg the stag representing
# a table construct in the FROM clause) and adds
# it to the alias hash if it specifies an alias
# ============================
sub get_table_alias_map {
    my $s = shift;
    my $h = shift;

    # the FROM clause is natively stored as a binary tree
    # (in order to group the joins by brackets) - recursively
    # descend building the hash map

    if ($s->name eq 'leaf') {
        my $alias = $s->get_alias;
        if ($alias) {
            $h->{$alias} = [$s->get_name];
        }
        return ($s->get_name);
    }
    elsif ($s->name eq 'composite') {
        my ($first, $second) =
          ($s->getnode_first,
           $s->getnode_second);
        my $alias = $s->get_alias;
        my @sn = ($first->subnodes, $second->subnodes);
        my @subtbls = map {
            get_table_alias_map($_, $h),
        } @sn;
        if ($alias) {
            $h->{$alias} = [@subtbls];
        }
        return @subtbls;
    }
    else {
        confess $s->name;
    }
}

# ============================
# reconstruct(schema, rows, top, cols, constraints, nesting, aliasstruct)
#
# mainly called by: selectall_stag(...)
#
# takes an array of rows (ie the result of an SQL query, probably
# involving JOINs, which is a denormalised relation) and
# decomposes this relation into a tree structure
#
# in order to do this, it requires schema information, and a nesting
# through the implicit result graph to build a tree
# ============================
sub reconstruct {
    my $self = shift;
    my $tree = Data::Stag->new();
    my ($schema,         # OPTIONAL - meta data on relation
        $rows,           # REQUIRED - relation R - array-of-array
        $top,            # OPTIONAL - root node name
        $cols,           # REQUIRED - array of stag nodes per column of R
        $constraints,    # NOT USED!!!
        $nesting,           # REQUIRED - tree representing decomposed schema
        $aliasstruct,    # OPTIONAL - renaming of columns in R
        $aliaspolicy) =
          rearrange([qw(schema
                        rows
                        top
                        cols
                        constraints
                        nesting
                        alias
                        aliaspolicy)], @_);

    $aliaspolicy = 'nest' unless $aliaspolicy;

    # --- get the schema ---
    #
    # $schema is a stag representing the schema
    # of the input releation R (not the schema of
    # the db that produced it.... hmm, this could
    # be misleading)
    #
    # it conforms to the following stag-struct:
    #
    #'(schema
    #  (top? "RECORDSET-ELEMENT-NAME")
    #  (cols?
    #   (col+
    #    (relation "RELATION-NAME")
    #    (name     "COLUMN-NAME")
    #    ))
    #  (nesting?
    #   (* "NESTING-TREE")))
    #
    # each column represents the

    if (!$schema) {
        $schema = $tree->new(schema=>[]);
    }
    if (!ref($schema)) {
        # it is a string - parse it
        # (assume sxpr)
        $schema = $tree->from('sxprstr', $schema);
    }

    # TOP - this is the element name
    # to group the structs under.
    # [override if specified explicitly]
    if ($top) {
        stag_set($schema, 'top', $top);
    }
#    $top = $schema->get_top || "set";
    if (!$top) {
	if ($nesting) {
	    # use first element in nesting
	    $top = $nesting->element;
	}
	else {
	    $top = 'set';
	}
    }
    my $topstruct = $tree->new($top, []);

    # COLS - this is the columns (attribute names)
    # in the order they appear
    # [override if specified explicitly]
    if ($cols) {
        my @ncols =
          map {
              if (ref($_)) {
                  $_
              }
              else {
                  # presume it's a string
                  # format = RELATION.ATTRIBUTENAME
                  if (/(\w+)\.(\w+)/) {
                      $tree->new(col=>[
                                       [relation=>$1],
                                       [name=>$2]]);
                  }
                  elsif (/(\w+)/) {
                      confess("Not implemented yet - must specify tbl for $_");
                      $tree->new(col=>[
                                       [relation=>'unknown'],
                                       [name=>$2]]);
                  }
                  else {
                      confess "I am confused by this column: $_";
                  }
              }
          } @$cols;
        $schema->set_cols([@ncols]);
    }


    # NESTING - this is the tree structure in
    # which the relations are structured
    # [override if specified explicitly]
    if ($nesting) {
        if (ref($nesting)) {
        }
        else {
            $nesting = $tree->from('sxprstr', $nesting);
        }
        $schema->set_nesting([$nesting]);
    }
    else {
        $nesting = $schema->sgetnode_nesting;
    }
    if (!$nesting) {
        confess("no nesting!");
    }

    # --- alias structure ---
    #
    # use this to get a hash map of alias => baserelation

    ($aliasstruct) = $schema->getnode_aliases unless $aliasstruct;
    if ($aliasstruct && !ref($aliasstruct)) {
        $aliasstruct = $tree->from('sxprstr', $aliasstruct);
    }
    my @aliases = ();
    if ($aliasstruct && $aliaspolicy !~ /^a/i) {
        @aliases = $aliasstruct->getnode_alias;
    }
    my %alias2baserelation =
      map {
          $_->sget_name => $_->sget_table
      } @aliases;

    # column headings; (ie all columns in R)
    my @cols = $schema->sgetnode_cols->getnode_col();

    # --- primary key info ---

    # set the primary key for each relation (one per relation);
    # the default is *all* the columns in that relation
    my %pkey_by_relationname = ();  # eg {person => [person_id]
    my %cols_by_relationname = ();  # eg {person => [person_id, fname, lname]

    # loop through all columns in R, setting above hash maps
    foreach my $col (@cols) {

	# the stag struct for each $col looks like this:
	#
	#   (col+
	#    (relation "RELATION-NAME")
	#    (name     "COLUMN-NAME")
	#    ))

        my $relationname = $col->get_relation;
        my $colname = $col->get_name;

	# pkey defaults to all columns in a relation
	# (we may override this later)
        $pkey_by_relationname{$relationname} = []
          unless $pkey_by_relationname{$relationname};
        push(@{$pkey_by_relationname{$relationname}},
             $colname);

	# all columns in a relation
	# (note: same as default PK)
        $cols_by_relationname{$relationname} = []
          unless $cols_by_relationname{$relationname};
        push(@{$cols_by_relationname{$relationname}},
             $colname);
    }
    my @relationnames = keys %pkey_by_relationname;

    # override PK if explicitly set as a constraint
    my @pks = $schema->findnode("primarykey");
    foreach my $pk (@pks) {

	# $pk looks like this:
	#
	# '(primarykey
	#   (relation "R-NAME")
	#   (col+ "COL-NAME"))

        my $relationname = $pk->get_relation;
        my @cols = $pk->get_col;

	# the hash %pkey_by_relationname should
	# be keyed by the named relations, not the
	# base relations
        my @aliasnames =
          grep {
              $alias2baserelation{$_} eq $relationname
          } keys %alias2baserelation;

        # relation is not aliased
        if (!@aliasnames) {
            @aliasnames = ($relationname);
        }
        foreach (@aliasnames) {
            $pkey_by_relationname{$_} = [@cols];
        }
    }

    # ------------------
    #
    # loop through denormalised rows,
    # putting the columns into their
    # respecive relations
    #
    # eg
    #
    #  <----- a ----->   <-- b -->
    #  a.1   a.2   a.3   b.1   b.2
    #
    # algorithm:
    #  use nesting/tree to walk through
    #
    # ------------------

    #~~~  keep a hash of all relations by their primary key vals
    #~~~   outer key = relationname
    #~~~   inner key = pkval
    #~~~   hash val  = relation structure
    #~~~ my %all_relation_hh = ();
    #~~~ foreach my $relationname (@relationnames) {
    #~~~     $all_relation_hh{$relationname} = {};
    #~~~ }

    #~~~ keep an array of all relations
    #~~~  outer key = relationname
    #~~~  inner array = ordered list of relations
    #~~~    my %all_relation_ah = ();
    #~~~    foreach my $relationname (keys %pkey_by_relationname) {
    #~~~        $all_relation_ah{$relationname} = [];
    #~~~    }

    # start at top of nesting tree
    #
    # a typical nesting tree may look like this:
    #
    # '(tableA
    #   (tableB "1")
    #   (tableC
    #    (tableD "1")))
    #
    # terminals ie "1" are ignored

    my ($first_in_nesting) = $nesting->subnodes;
    if (!$first_in_nesting) {
        $first_in_nesting = $nesting;
    }
    my $fipname = $first_in_nesting ? $first_in_nesting->name : '';

    # recursive hash representing tree
    #
    # $record =
    #   {child_h => {
    #                $relation_name* => {
    #                                   $pk_val => $record
    #                                  }
    #               },
    #    struct => $stag_obj
    #   }
    #
    # this is recursively constructed using the make_a_tree() method
    # below. the nesting tree (see above) is traversed depth first,
    # constructing both the child_h hash and the resulting Stag
    # structure.

    my $top_record_h =
      {
       child_h=>{ $fipname ? ($fipname=>{}) : () },
       struct=>$topstruct
      };
    # loop through rows in R
    foreach my $row (@$rows) {
        my @colvals = @$row;

        # keep a record of all table names in
        # this row from R
        my %current_relation_h = ();
        for (my $i=0; $i<@cols; $i++) {
            my $colval = $colvals[$i];
            my $col = $cols[$i];
            my $relationname = $col->get_relation;
            my $colname = $col->get_name;
            my $relation = $current_relation_h{$relationname};
            if (!$relation) {
                $relation = {};
                $current_relation_h{$relationname} = $relation;
            }
            $relation->{$colname} = $colval;
        }

#	print "ROW=@$row\n";
#	dmp(\%pkey_by_relationname);
#	dmp($top_record_h);

        # we now have a hash of hashes -
        #  outer keyed by relation id
        #  inner keyed by relation attribute name

        # traverse depth first down nesting;
        # add new nodes as children of the parent
        $self->make_a_tree($tree,
                           $top_record_h,
                           $first_in_nesting,
                           \%current_relation_h,
                           \%pkey_by_relationname,
                           \%cols_by_relationname,
                           \%alias2baserelation,
                           $aliaspolicy);
    }
    return $topstruct;
}
*norm = \&reconstruct;
*normalise = \&reconstruct;
*normalize = \&reconstruct;

# ============================
# make_a_tree(...) RECURSIVE
#
# called by: reconstruct(...)
#
# ============================
sub make_a_tree {
    my $self = shift;
    my $tree = shift;
    my $parent_rec_h = shift;
    my $node = shift;
    my %current_relation_h= %{shift ||{}};
    my %pkey_by_relationname = %{shift ||{}};
    my %cols_by_relationname = %{shift ||{}};
    my %alias2baserelation = %{shift ||{}};
    my $aliaspolicy = shift;

    my $relationname = $node->name;
    my $relationrec = $current_relation_h{$relationname};
    my $pkcols = $pkey_by_relationname{$relationname};
    my $rec; # this is the next node down in the hash tree

    if (!$pkcols || !@$pkcols) {
	# if we have no columns for a particular part of
	# the nesting through the relation, it means it
	# was ommitted from the select clause - just skip
	# this part of the nesting.
	#
	# for example: SELECT a.*, b.* FROM a NJ a_to_b NJ b
	# the default nesting will be: [a [a_to_b [b]]]
	# the relation R will have columns:
	# a.c1    a.c2    b.c1   b.c2
	#
	# we want to build a resulting structure like this:
	# (a
	#  (c1 "x") (c2 "y")
	#  (b
	#   (c1 "a") (c2 "b")))
	#
	# so we just miss out a_to_b in the nesting, because it
	# has no columns in the relation R.
	$rec = $parent_rec_h;
    }
    else {

	my $pkval =
	  CORE::join("\t",
		     map {
			 esctab($relationrec->{$_} || '')
		     } @$pkcols);

	$rec = $parent_rec_h->{child_h}->{$relationname}->{$pkval};

	if (!$rec) {
	    my $relationcols = $cols_by_relationname{$relationname};
	    my $has_non_null_val = grep {defined($relationrec->{$_})} @$relationcols;
	    return unless $has_non_null_val;
	    my $relationstruct =
	      $tree->new($relationname=>[
					 map {
					     defined($relationrec->{$_}) ? [$_ => $relationrec->{$_}] : ()
					 } @$relationcols
					]);
	    my $parent_relationstruct = $parent_rec_h->{struct};
	    if (!$parent_relationstruct) {
		confess("no parent for $relationname");
	    }

	    # if we have an aliased relation, add an extra
	    # level of nesting
	    my $baserelation = $alias2baserelation{$relationname};
	    if ($baserelation) {

                # $aliaspolicy eq 'nest' or 't'
                # nest base relations inside an alias node
                # OR use table name in place of alias name
                if ($aliaspolicy =~ /^t/i) {
                    stag_add($parent_relationstruct,
                             $baserelation,
                             $relationstruct->data);
                }
                else {
                    # nest
                    my $baserelationstruct =
                      Data::Stag->new($baserelation =>
                                      $relationstruct->data);
                    stag_add($parent_relationstruct,
                             $relationname,
                             [$baserelationstruct]);
                }
	    } else {
                # either no aliases, or $aliaspolicy eq 'a'
                # (in which case columns already mapped to aliases)
                stag_add($parent_relationstruct,
                         $relationstruct->name,
                         $relationstruct->data);
	    }
	    $rec =
	      {struct=>$relationstruct,
	       child_h=>{}};
	    foreach ($node->subnodes) {
		# keep index of children by PK
		$rec->{child_h}->{$_->name} = {};
	    }
	    $parent_rec_h->{child_h}->{$relationname}->{$pkval} = $rec;
	}
    }
    foreach ($node->subnodes) {
        $self->make_a_tree($tree,
			   $rec,
			   $_,
			   \%current_relation_h,
			   \%pkey_by_relationname,
			   \%cols_by_relationname,
			   \%alias2baserelation,
                           $aliaspolicy);
    }
}


# -------- GENERAL SUBS -----------

sub esctab {
    my $w=shift;
    $w =~ s/\t/__MAGICTAB__/g;
    $w;
}

sub makesql {
    my $self = shift;
    my ($table,
        $where,
        $select,
        $order,
        $group,
        $distinct) =
          rearrange([qw(table
                        where
                        select
                        order
                        group
                        distinct)], @_);

    confess("must specify table") unless $table;

    # array of tables
    if (ref($table)) {
        if (ref($table) eq "HASH") {
            $table =
              [
               map {
                   "$table->{$_} AS $_"
               } keys %$table
              ];
        }
    }
    else {
        $table = [$table];
    }

    $where = [] unless $where;
    # array of ANDed where clauses
    if (ref($where)) {
        if (ref($where) eq "HASH") {
            $where =
              [
               map {
                   "$_ = ".$self->quote($where->{$_})
               } keys %$where
              ];
        }
    }
    else {
        $where = [$where];
    }

    $select = ['*'] unless $select;
    # array of SELECT cols
    if (ref($select)) {
        if (ref($select) eq "HASH") {
            $select =
              [
               map {
                   "$select->{$_} AS $_"
               } keys %$select
              ];
        }
    }
    else {
        $select = [$select];
    }

    $order = [] unless $order;
    # array of order tables
    if (ref($order)) {
        if (ref($order) eq "HASH") {
            confess("order must be an array");
        }
    }
    else {
        $order = [$order];
    }

    $group = [] unless $group;
    # array of group tables
    if (ref($group)) {
        if (ref($group) eq "HASH") {
            confess("group must be an array");
        }
    }
    else {
        $group = [$group];
    }

    $distinct = $distinct ? '' : ' DISTINCT';
    my $sql =
      sprintf("SELECT%s %s FROM %s%s%s",
              $distinct,
              join(', ', @$select),
              join(', ', @$table),
              (scalar(@$where) ?
               ' WHERE '.join(' AND ', @$where) : ''),
              (scalar(@$group) ?
               ' GROUP BY '.join(', ', @$group) : ''),
              (scalar(@$order) ?
               ' ORDER BY '.join(', ', @$order) : ''),
             );
    return $sql;
}



sub selectval {
    my $self = shift;
    trace(0, "@_") if $TRACE;
    return $self->dbh->selectcol_arrayref(@_)->[0];
}

sub insertrow {
    my $self = shift;
    my ($table, $colvalh, $pkcol) = @_;

    my $driver = $self->dbh->{Driver}->{Name};
    my @cols = keys %$colvalh;
    my @vals =
      map {
          defined($_) ? $colvalh->{$_} : undef
      } @cols;
    my @placeholders = map { '?' } @vals;
    my $sql =
      sprintf("INSERT INTO %s (%s) VALUES (%s)",
              $table,
              join(", ", @cols),
              #join(", ", @vals),
              join(", ", @placeholders),
             );
    if (!@cols) {
	$sql = "INSERT INTO $table DEFAULT VALUES";
    }

    trace(0, "SQL:$sql") if $TRACE;
    my $succeeded = 0;
    eval {
        my $sth = $self->dbh->prepare($sql);
        my $rval = $sth->execute(@vals);
        $succeeded = 1 if defined $rval;
    };
    if ($@) {
	if ($self->force) {
	    # what about transactions??
	    $self->warn("IN SQL: $sql\nWARNING: $@\n");
            return;
	}
	else {
	    confess $@;
	}
    }
    return unless $succeeded;
    my $pkval;
    if ($pkcol) {
        # primary key value may have been specified in the xml
        # (this is necessary for non-surrogate pks in tables that
        #  are to be linked to via foreign keys)
        $pkval = $colvalh->{$pkcol};

        # pk was not supplied - perhaps this is a SERIAL/AUTO_INCREMENT
        # column (ie surrogate integer primary key)
        if (!$pkval) {
            # assume pk is a SERIAL / AUTO_INCREMENT
            if ($driver eq 'Pg') {
                my $seqn = sprintf("%s_%s_seq",
                                   $table,
                                   $pkcol);
                $pkval  = $self->selectval("select currval('$seqn')");
                trace(0, "CURRVAL $seqn = $pkval     [Pg]") if $TRACE;
            }
            # this doesn't work on older
            # versions of DBI/DBD::mysql
            # seems to have been fixed Oct 2004 release
            elsif ($driver eq 'mysql') {
                $pkval = $self->dbh->last_insert_id(undef,undef,$table,$pkcol);
                trace(0, "CURRVAL mysql_insert_id $pkval   [mysql]") if $TRACE;
            }
            else {
                $pkval  = $self->selectval("select max($pkcol) from $table");
            }
        }
        trace(0, "PKVAL = $pkval") if $TRACE;
    }
    return $pkval;
}

sub updaterow {
    my $self = shift;
    my ($table, $set, $where) = @_;

    confess("must specify table") unless $table;

    my $dbh = $self->dbh;

    # array of WHERE cols
    if (ref($where)) {
        if (ref($where) eq "HASH") {
            $where =
              [
               map {
                   "$_ = ".$dbh->quote($where->{$_})
               } keys %$where
              ];
        }
    }
    else {
        $where = [$where];
    }
    confess("must specify constraints") unless @$where;

    confess("must set update vals") unless $set;
    my @bind = ();
    # array of SET colvals
    if (ref($set)) {
        if (ref($set) eq "HASH") {
            $set =
              [
               map {
                   push(@bind, defined $set->{$_} ? $set->{$_} : 'NULL');
                   "$_ = ?"
               } keys %$set
              ];
        }
    }
    else {
        $set = [$set];
    }

    my $sql =
      sprintf("UPDATE %s SET %s WHERE %s",
              $table,
              join(', ', @$set),
              join(' AND ', @$where),
             );
    trace(0, "SQL:$sql [",join(', ',@bind)."]") if $TRACE;

    my $sth = $dbh->prepare($sql) || confess($sql."\n\t".$dbh->errstr);
    return $sth->execute(@bind) || confess($sql."\n\t".$sth->errstr);
}

sub deleterow {
    my $self = shift;
    my ($table, $where) = @_;

    confess("must specify table") unless $table;

    my $dbh = $self->dbh;

    # array of WHERE cols
    if (ref($where)) {
        if (ref($where) eq "HASH") {
            $where =
              [
               map {
                   "$_ = ".$dbh->quote($where->{$_})
               } keys %$where
              ];
        }
    }
    else {
        $where = [$where];
    }
    confess("must specify constraints") unless @$where;

    my $sql =
      sprintf("DELETE FROM %s WHERE %s",
              $table,
              join(' AND ', @$where),
             );
    trace(0, "SQL:$sql") if $TRACE;

    my $sth = $dbh->prepare($sql) || confess($sql."\n\t".$dbh->errstr);
    return $sth->execute() || confess($sql."\n\t".$sth->errstr);
}

#$::RD_HINT = 1;

$::RD_AUTOACTION = q { [@item] };
sub selectgrammar {
    return q[

             {
              use Data::Dumper;
              use Data::Stag;
              sub N {
                  Data::Stag->new(@_);
              }
          }
         ]
       .
         q[

         selectstmts: selectstmt ';' selectstmts
         selectstmts: selectstmt
           #           selectstmt: /select/i selectcols /from/i fromtables
         selectstmt: /select/i selectq(?) selectcols /from/i fromtables where(?) group(?) having(?) combiner(?) order(?) limit(?) offset(?)
           {
               N(select => [
                            [qual => $item{'selectq'}[0]],
                            [cols => $item[3]],
                            [from => $item[5]],
#                            [where => $item[6]],
#                            [group => $item{'group'}[0]],
#                            [having => $item{'having'}[0]],
                           ]);
           }
           |
         selectq: /all/i | /distinct/i
               { $item[1] }
           |
#	 as: /\s+as\s+/i
	 as: /as/i
         selectcols: selectexpr /\,/ selectcols
           { [$item[1], @{$item[3]}] }
           |
         selectcols: selectexpr
             { [$item[1]] }
           |
         selectexpr: bselectexpr as aliasname
           {
               my $col = $item{bselectexpr};
               $col->set_alias($item{aliasname}->[1]);
               $col;
           }
           |
	 selectexpr: bselectexpr
            { $item[1] }
            |
         bselectexpr: funccall
           { $item[1] }
           |
         bselectexpr: selectcol
           { $item[1] }
           |

	 selectcol: brackselectcol operator selectcol
	   {
	       N(col=>[
		       [func => [
				 [name => $item[2]->[1]],
				 [args => [$item[1],$item[3]]]
				]
		       ]
		      ]);
	   }
###	   { $item[1]}
	   |
	 selectcol: brackselectcol
	   { $item[1]}
	   |

	 brackselectcol: '(' selectcol ')'
	   { $item[2]}
	   |

	 brackselectcol: bselectcol
	   { $item[1]}
	   |

         bselectcol: /(\w+)\.(\w+)/
           { N(col=>[
                     [name => $item[1]],
		     [table=>$1],
                    ])
           }
           |
         bselectcol: /(\w+)\.\*/
           { N(col=>[
                     [name => $item[1]],
                     [table=>$1],
                    ])
           }
           |
         bselectcol: /\*/
           { N(col=>[
                     [name => $item[1]]
                    ])
           }
           |
         bselectcol: /\w+/
           { N(col=>[
                     [name => $item[1]]
                    ])
           }
           |
         bselectcol: expr
           { N(col=>[
                     [expr => $item[1]]
                    ]) }
           |
         funccall: funcname '(' distinct(?) selectcols ')'
           {
            my $col = N(col=>[
                              [func => [
                                        [name => $item[1]->[1]],
                                        [args => $item[4]]
                                       ]
                              ]
                             ]);
            $col;
           }
           |

         distinct: /distinct/i

	 operator: '+' | '-' | '*' | '/' | '||'


         fromtables: jtable
           { [$item[1]] }
           |
         jtable: join_jtable
           { $item[1] }
           |
         join_jtable: qual_jtable jointype join_jtable
           {
               shift @{$item[2]};
               my $j =
                 N(composite=>[
                               [ctype=>"@{$item[2]}"],
                               [first=>[$item[1]]],
                               [second=>[$item[3]]]
                              ]);
               $j;
         }
           |
         join_jtable: qual_jtable
           { $item[1] }
           |
         qual_jtable: alias_jtable joinqual
           {
               my $j = $item[1];
               $j->setnode_qual($item[2]);
               $j;
           }
           |
         qual_jtable: alias_jtable
           { $item[1] }
           |
         alias_jtable: brack_jtable /as\s+/i aliasname
           {
               my $j = $item[1];
               $j->set_alias($item[3][1]);
               $j;
           }
           |
         alias_jtable: brack_jtable
           { $item[1] }
           |
         brack_jtable: '(' jtable ')'
           { $item[2] }
           |
         brack_jtable: table
           { N(leaf=>[[name=>$item[1]->[1]]]) }
           |

         joinqual: /on\s+/i bool_expr
           { N(qual => [
                        [type=>'on'],
                        [expr=>"@{$item[2]}"]
                       ])
         }
           |
         joinqual: /using\s+/i '(' cols ')'
           { N(qual =>[
                       [type=>'using'],
                       [expr=>"@{$item[3]}"]
                      ])
         }
           |

         table: tablename
           { $item[1] }
           |

         funcname: /\w+/
         tablename: /\w+/
         aliasname: /\w+/


         cols: col(s)
         col: /\w+\.\w+/
         col: /\w+/

         jointype: /\,/
         jointype: /natural/i bjointype /join/i
         jointype: /natural/i /join/i
         jointype: bjointype /join/i
         jointype: /join/i
         bjointype: /inner/i
         bjointype: lrf(?) /outer/i
         lrf: /left/i | /right/i | /full/i
         bjointype: /cross/i

         number: float | int
         float: /\d*\.?\d+/ 'e' sign int
         float: /\d*\.\d+/
         int: /\d+/
         string: /\'.*?\'/
         sign: '+' | '-'

         exprs: '(' exprs ')'
         exprs: expr ',' exprs
         exprs: expr

	   # bool_expr - eg in where clause
         bool_expr: not_bool_expr boolop bool_expr | not_bool_expr
         not_bool_expr: '!' brack_bool_expr | brack_bool_expr
         brack_bool_expr: '(' bool_expr ')' | bool_exprprim
         bool_exprprim: boolval | expr
         boolval: /true/i | /false/i | /null/i

         expr: brack_expr op expr | brack_expr
         brack_expr: '(' expr ')' | exprprim
         exprprim: col | val
         val: number | string

         op: /not\s+/i /like\s+/i
         op: /like\s+/i
         op: /is\s+/i /not\s+/i
         op: /is\s+/i
         op: '=' | '!=' | '<>' | '<=' | '>=' | '<' | '>'
         boolop: /and\s+/i | /or\s+/i | /not\s+/i

#           where: /where/i /.*/
           where: /where/i bool_expr
           group: /group/i /by/i exprs
           having: /having/i /.*/
           combiner: combinekwd selectstmt
           combinekwd: /union/i | /intersect/i | /update/i
           order: /order/i /by/i orderexprs
           orderexprs: orderexpr ',' orderexprs
           orderexprs: orderexpr
           orderexpr: expr /asc/i
           orderexpr: expr /desc/i
           orderexpr: expr /using/i op
           orderexpr: expr
           limit: /limit/i /\w+/
           offset: /offset/i /\d+/
            ];
}

no strict 'refs';
sub AUTOLOAD {
    my $self = shift;
    my @args = @_;

    my $name = $AUTOLOAD;
    $name =~ s/.*://;   # strip fully-qualified portion

    if ($name eq "DESTROY") {
	# we dont want to propagate this!!
	return;
    }

    unless ($self->isa("DBIx::DBStag")) {
        confess("no such subroutine $name");
    }
    if ($self->dbh) {
	if ($TRACE) {
	    # the following check may impair performance
	    if (grep { ref($_) } @args) {
		$self->throw("cannot quote @args");
	    }
	}
	if ($self->dbh->can($name)) {
	    return $self->dbh->$name(@args);
	}
    }
    confess("no such method:$name)");
}

sub rearrange {
  my($order,@param) = @_;

  # If there are no parameters, we simply wish to return
  # an undef array which is the size of the @{$order} array.
  return (undef) x $#{$order} unless @param;

  # If we've got parameters, we need to check to see whether
  # they are named or simply listed. If they are listed, we
  # can just return them.
  return @param unless (defined($param[0]) && $param[0]=~/^-/);

  # Now we've got to do some work on the named parameters.
  # The next few lines strip out the '-' characters which
  # preceed the keys, and capitalizes them.
  my $i;
  for ($i=0;$i<@param;$i+=2) {
      if (!defined($param[$i])) {
	  cluck("Hmmm in $i ".CORE::join(";", @param)." == ".CORE::join(";",@$order)."\n");
      }
      else {
	  $param[$i]=~s/^\-//;
	  $param[$i]=~tr/a-z/A-Z/;
      }
  }

  # Now we'll convert the @params variable into an associative array.
  my(%param) = @param;

  my(@return_array);

  # What we intend to do is loop through the @{$order} variable,
  # and for each value, we use that as a key into our associative
  # array, pushing the value at that key onto our return array.
  my($key);

  foreach $key (@{$order}) {
      $key=~tr/a-z/A-Z/;
      my($value) = $param{$key};
      delete $param{$key};
      push(@return_array,$value);
  }

  # catch user misspellings resulting in unrecognized names
  my(@restkeys) = keys %param;
  if (scalar(@restkeys) > 0) {
       confess("@restkeys not processed in rearrange(), did you use a
       non-recognized parameter name ? ");
  }
  return @return_array;
}

#sub loadschema {
#    my $self = shift;
#    my ($ddl, $ddlf, $dialect) =
#      rearrange([qw(ddl ddlf dialect)], @_);
#    if ($ddlf) {
#        my $fh = FileHandle->new($ddlf) || $self->throw("no file $ddlf");
#        $ddl = join('',<$fh>);
#        $fh->close;
#    }
#    $self->throw("no DDL") unless $ddl;
#    if ($dialect) {
#        my $driver = $self->{_driver} || 'Pg';
#        if ($driver ne $dialect) {

#        }
#    }
#}

1;

__END__