File Coverage

blib/lib/SQL/Abstract/More.pm

Criterion	Covered	Total	%
statement	513	579	88.6
branch	210	280	75.0
condition	57	95	60.0
subroutine	64	74	86.4
pod	13	13	100.0
total	857	1041	82.3

package SQL::Abstract::More;
use strict;
use warnings;

# no "use parent ..." here -- the inheritance is specified dynamically in the
# import() method -- inheriting either from SQL::Abstract or SQL::Abstract::Classic

use MRO::Compat;
use mro 'c3'; # implements next::method

use Params::Validate  qw/validate SCALAR SCALARREF CODEREF ARRAYREF HASHREF
                                  UNDEF  BOOLEAN/;
use Scalar::Util      qw/blessed reftype/;


# remove all previously defined or imported functions
use namespace::clean;

# declare error-reporting functions from SQL::Abstract
sub puke(@); sub belch(@);  # these will be defined later in import()

our $VERSION = '1.39';
our @ISA;

sub import {
  my $class = shift;

  # parent class specified from environment variable, or default value
  my $parent_sqla = $ENV{SQL_ABSTRACT_MORE_EXTENDS} || 'SQL::Abstract::Classic';

  # parent class specified through -extends => .. when calling import()
  $parent_sqla = $_[1] if @_ >= 2 && $_[0] eq '-extends';

  # syntactic sugar : 'Classic' is expanded into SQLA::Classic
  $parent_sqla = 'SQL::Abstract::Classic' if $parent_sqla eq 'Classic';

  # make sure that import() does never get called with different parents
  if (my $already_isa = $ISA[0]) {
    $already_isa eq $parent_sqla
      or die "cannot use SQL::Abstract::More -extends => '$parent_sqla', "
           . "this module was already loaded with -extends => '$already_isa'";

    # the rest of the import() job was already performed, so just return from here
    return;
  }

  # load the parent, inherit from it, import puke() and belch()
  eval qq{use parent '$parent_sqla';
          *puke  = \\&${parent_sqla}::puke;
          *belch = \\&${parent_sqla}::belch;
         };

  # local override of some methods for insert() and update()
  _setup_insert_inheritance($parent_sqla);
  _setup_update_inheritance($parent_sqla);
}



#----------------------------------------------------------------------
# Utility functions -- not methods -- declared _after_
# namespace::clean so that they can remain visible by external
# modules. In particular, DBIx::DataModel imports these functions.
#----------------------------------------------------------------------

# shallow_clone(): copies of the top-level keys and values, blessed into the same class
sub shallow_clone {
  my ($orig, %override) = @_;

  my $class = ref $orig
    or puke "arg must be an object";
  my $clone = {%$orig, %override};
  return bless $clone, $class;
}


# does(): cheap version of Scalar::Does
my %meth_for = (
  ARRAY  => '@{}',
  HASH   => '%{}',
  SCALAR => '${}',
  CODE   => '&{}',
 );
sub does ($$) {
  my ($data, $type) = @_;
  my $reft = reftype $data;
  return defined $reft && $reft eq $type
      || blessed $data && overload::Method($data, $meth_for{$type});
}



#----------------------------------------------------------------------
# global variables
#----------------------------------------------------------------------

# builtin methods for "Limit-Offset" dialects
my %limit_offset_dialects = (
  LimitOffset => sub {my ($self, $limit, $offset) = @_;
                      $offset ||= 0;
                      return "LIMIT ? OFFSET ?", $limit, $offset;},
  LimitXY     => sub {my ($self, $limit, $offset) = @_;
                      $offset ||= 0;
                      return "LIMIT ?, ?", $offset, $limit;},
  LimitYX     => sub {my ($self, $limit, $offset) = @_;
                      $offset ||= 0;
                      return "LIMIT ?, ?", $limit, $offset;},
  RowNum      => sub {
    my ($self, $limit, $offset) = @_;
    # HACK below borrowed from SQL::Abstract::Limit. Not perfect, though,
    # because it brings back an additional column. Should borrow from
    # DBIx::Class::SQLMaker::LimitDialects, which does the proper job ...
    # but it says : "!!! THIS IS ALSO HORRIFIC !!! /me ashamed"; so
    # I'll only take it as last resort; still exploring other ways.
    # See also L : within that ORM an additional layer is
    # added to take advantage of Oracle scrollable cursors.
    my $sql = "SELECT * FROM ("
            .   "SELECT subq_A.*, ROWNUM rownum__index FROM (%s) subq_A "
            .   "WHERE ROWNUM <= ?"
            .  ") subq_B WHERE rownum__index >= ?";

    no warnings 'uninitialized'; # in case $limit or $offset is undef
    # row numbers start at 1
    return $sql, $offset + $limit, $offset + 1;
  },
 );

# builtin join operators with associated sprintf syntax
my %common_join_syntax = (
  '<=>' => '%s INNER JOIN %s ON %s',
   '=>' => '%s LEFT OUTER JOIN %s ON %s',
  '<='  => '%s RIGHT OUTER JOIN %s ON %s',
  '=='  => '%s NATURAL JOIN %s',
  '>=<' => '%s FULL OUTER JOIN %s ON %s',
);
my %right_assoc_join_syntax = %common_join_syntax;
s/JOIN %s/JOIN (%s)/ foreach values %right_assoc_join_syntax;

# specification of parameters accepted by the new() method
my %params_for_new = (
  table_alias          => {type => SCALAR|CODEREF,   default  => '%s AS %s'},
  column_alias         => {type => SCALAR|CODEREF,   default  => '%s AS %s'},
  limit_offset         => {type => SCALAR|CODEREF,   default  => 'LimitOffset'},
  join_syntax          => {type => HASHREF,          default  =>
                                                        \%common_join_syntax},
  join_assoc_right     => {type => BOOLEAN,          default  => 0},
  max_members_IN       => {type => SCALAR,           optional => 1},
  multicols_sep        => {type => SCALAR|SCALARREF, optional => 1},
  has_multicols_in_SQL => {type => BOOLEAN,          optional => 1},
  sql_dialect          => {type => SCALAR,           optional => 1},
  select_implicitly_for=> {type => SCALAR|UNDEF,     optional => 1},
);

# builtin collection of parameters, for various databases
my %sql_dialects = (
 MsAccess  => { join_assoc_right     => 1,
                join_syntax          => \%right_assoc_join_syntax},
 BasisJDBC => { column_alias         => "%s %s",
                max_members_IN       => 255                      },
 MySQL_old => { limit_offset         => "LimitXY"                },
 Oracle    => { limit_offset         => "RowNum",
                max_members_IN       => 999,
                table_alias          => '%s %s',
                column_alias         => '%s %s',
                has_multicols_in_SQL => 1,                       },
);


# operators for compound queries
my @set_operators = qw/union union_all intersect minus except/;

# specification of parameters accepted by select, insert, update, delete
my %params_for_select = (
  -columns      => {type => SCALAR|ARRAYREF,         default  => '*'},
  -from         => {type => SCALAR|SCALARREF|ARRAYREF},
  -where        => {type => SCALAR|ARRAYREF|HASHREF, optional => 1},
  (map {-$_ => {type => ARRAYREF, optional => 1}} @set_operators),
  -group_by     => {type => SCALAR|ARRAYREF,         optional => 1},
  -having       => {type => SCALAR|ARRAYREF|HASHREF, optional => 1},
  -order_by     => {type => SCALAR|ARRAYREF|HASHREF, optional => 1},
  -page_size    => {type => SCALAR,                  optional => 1},
  -page_index   => {type => SCALAR,                  optional => 1,
                                                     depends  => '-page_size'},
  -limit        => {type => SCALAR,                  optional => 1},
  -offset       => {type => SCALAR,                  optional => 1,
                                                     depends  => '-limit'},
  -for          => {type => SCALAR|UNDEF,            optional => 1},
  -want_details => {type => BOOLEAN,                 optional => 1},
);
my %params_for_insert = (
  -into         => {type => SCALAR},
  -values       => {type => SCALAR|ARRAYREF|HASHREF, optional => 1},
  -select       => {type => HASHREF,                 optional => 1},
  -columns      => {type => ARRAYREF,                optional => 1},
  -returning    => {type => SCALAR|ARRAYREF|HASHREF, optional => 1},
  -add_sql      => {type => SCALAR,                  optional => 1},
);
my %params_for_update = (
  -table        => {type => SCALAR|SCALARREF|ARRAYREF},
  -set          => {type => HASHREF},
  -where        => {type => SCALAR|ARRAYREF|HASHREF, optional => 1},
  -order_by     => {type => SCALAR|ARRAYREF|HASHREF, optional => 1},
  -limit        => {type => SCALAR,                  optional => 1},
  -returning    => {type => SCALAR|ARRAYREF|HASHREF, optional => 1},
  -add_sql      => {type => SCALAR,                  optional => 1},
);
my %params_for_delete = (
  -from         => {type => SCALAR},
  -where        => {type => SCALAR|ARRAYREF|HASHREF, optional => 1},
  -order_by     => {type => SCALAR|ARRAYREF|HASHREF, optional => 1},
  -limit        => {type => SCALAR,                  optional => 1},
  -add_sql      => {type => SCALAR,                  optional => 1},
);
my %params_for_WITH = (
  -table        => {type => SCALAR},
  -columns      => {type => SCALAR|ARRAYREF,         optional => 1},
  -as_select    => {type => HASHREF},
  -final_clause => {type => SCALAR,                  optional => 1},
);



#----------------------------------------------------------------------
# object creation
#----------------------------------------------------------------------

sub new {
  my $class = shift;
  my %params = does($_[0], 'HASH') ? %{$_[0]} : @_;

  # extract params for this subclass
  my %more_params;
  foreach my $key (keys %params_for_new) {
    $more_params{$key} = delete $params{$key} if exists $params{$key};
  }

  # import params from SQL dialect, if any
  my $dialect = delete $more_params{sql_dialect};
  if ($dialect) {
    my $dialect_params = $sql_dialects{$dialect}
      or puke "no such sql dialect: $dialect";
    $more_params{$_} ||= $dialect_params->{$_} foreach keys %$dialect_params;
  }

  # check parameters for this class
  my @more_params = %more_params;
  my $more_self   = validate(@more_params, \%params_for_new);

  # check some of the params for parent -- because SQLA doesn't do it :-(
  !$params{quote_char} || exists $params{name_sep}
    or belch "when 'quote_char' is present, 'name_sep' should be present too";

  # call parent constructor
  my $self = $class->next::method(%params);

  # inject into $self
  $self->{$_} = $more_self->{$_} foreach keys %$more_self;

  # arguments supplied as scalars are transformed into coderefs
  ref $self->{column_alias} or $self->_make_sub_column_alias;
  ref $self->{table_alias}  or $self->_make_sub_table_alias;
  ref $self->{limit_offset} or $self->_choose_LIMIT_OFFSET_dialect;

  # regex for parsing join specifications
  my @join_ops = sort {length($b) <=> length($a) || $a cmp $b}
                      keys %{$self->{join_syntax}};
  my $joined_ops = join '|', map quotemeta, @join_ops;
  $self->{join_regex} = qr[
     ^              # initial anchor
     ($joined_ops)? # $1: join operator (i.e. '<=>', '=>', etc.))
     ([[{])?        # $2: opening '[' or '{'
     (.*?)          # $3: content of brackets
     []}]?          # closing ']' or '}'
     $              # final anchor
   ]x;

  return $self;
}



#----------------------------------------------------------------------
# support for WITH or WITH RECURSIVE
#----------------------------------------------------------------------

sub with_recursive {
  my $self = shift;

  my $new_instance = $self->with(@_);
  $new_instance->{WITH}{sql} =~ s/^WITH\b/WITH RECURSIVE/;

  return $new_instance;
}

sub with {
  my $self = shift;

  ! $self->{WITH}
    or puke "calls to the with() or with_recursive() method cannot be chained";

  @_
    or puke "->with() : missing arguments";

  # create a copy of the current object with an additional attribute WITH
  my $clone = shallow_clone($self, WITH => {sql => "", bind => []});

  # assemble SQL and bind values for each table expression
  my @table_expressions = does($_[0], 'ARRAY') ? @_ : ( [ @_]);
  foreach my $table_expression (@table_expressions) {
    my %args = validate(@$table_expression, \%params_for_WITH);
    my ($sql, @bind) = $self->select(%{$args{-as_select}});
    $clone->{WITH}{sql} .= ", " if $clone->{WITH}{sql};
    $clone->{WITH}{sql} .= $args{-table};
    $clone->{WITH}{sql} .= "(" . join(", ", @{$args{-columns}}) . ")" if $args{-columns};
    $clone->{WITH}{sql} .= " AS ($sql) ";
    $clone->{WITH}{sql} .= $args{-final_clause} . " "                 if $args{-final_clause};
    push @{$clone->{WITH}{bind}}, @bind;
  }

  # add the initial keyword WITH
  substr($clone->{WITH}{sql}, 0, 0, 'WITH ');

  return $clone;
}


sub _prepend_WITH_clause {
  my ($self, $ref_sql, $ref_bind) = @_;

  return if !$self->{WITH};

  substr($$ref_sql, 0, 0, $self->{WITH}{sql});
  unshift @$ref_bind, @{$self->{WITH}{bind}};

}


#----------------------------------------------------------------------
# the select method
#----------------------------------------------------------------------

sub select {
  my $self = shift;

  # if got positional args, this is not our job, just delegate to the parent
  return $self->next::method(@_) if !&_called_with_named_args;

  my %aliased_columns;

  # parse arguments
  my %args = validate(@_, \%params_for_select);

  # compute join info if the datasource is a join
  my $join_info = $self->_compute_join_info($args{-from});
  $args{-from}  = \($join_info->{sql}) if $join_info;

  # reorganize columns; initial members starting with "-" are extracted
  # into a separate list @post_select, later re-injected into the SQL
  my @cols = ref $args{-columns} ? @{$args{-columns}} : $args{-columns};
  my @post_select;
  push @post_select, shift @cols while @cols && $cols[0] =~ s/^-//;
  foreach my $col (@cols) {
    # extract alias, if any
    if ($col =~ /^\s*         # ignore insignificant leading spaces
                 (.*[^|\s])   # any non-empty string, not ending with ' ' or '|'
                 \|           # followed by a literal '|'
                 (\w+)        # followed by a word (the alias))
                 \s*          # ignore insignificant trailing spaces
                 $/x) {
      $aliased_columns{$2} = $1;
      $col = $self->column_alias($1, $2);
    }
  }
  $args{-columns} = \@cols;

  # reorganize pagination
  if ($args{-page_index} || $args{-page_size}) {
    not exists $args{$_} or puke "-page_size conflicts with $_"
      for qw/-limit -offset/;
    $args{-limit} = $args{-page_size};
    if ($args{-page_index}) {
      $args{-offset} = ($args{-page_index} - 1) * $args{-page_size};
    }
  }

  # generate initial ($sql, @bind), without -order_by (will be handled later)
  my @old_API_args = @args{qw/-from -columns -where/}; #
  my ($sql, @bind) = $self->next::method(@old_API_args);
  unshift @bind, @{$join_info->{bind}} if $join_info;

  # add @post_select clauses if needed (for ex. -distinct)
  my $post_select = join " ", @post_select;
  $sql =~ s[^SELECT ][SELECT $post_select ]i if $post_select;

  # add set operators (UNION, INTERSECT, etc) if needed
  foreach my $set_op (@set_operators) {
    if ($args{-$set_op}) {
      my %sub_args = @{$args{-$set_op}};
      $sub_args{$_} ||= $args{$_} for qw/-columns -from/;
      local $self->{WITH}; # temporarily disable the WITH part during the subquery
      my ($sql1, @bind1) = $self->select(%sub_args);
      (my $sql_op = uc($set_op)) =~ s/_/ /g;
      $sql .= " $sql_op $sql1";
      push @bind, @bind1;
    }
  }

  # add GROUP BY if needed
  if ($args{-group_by}) {
    my $sql_grp = $self->where(undef, $args{-group_by});
    $sql_grp =~ s/\bORDER\b/GROUP/;
    $sql .= $sql_grp;
  }

  # add HAVING if needed (often together with -group_by, but not always)
  if ($args{-having}) {
    my ($sql_having, @bind_having) = $self->where($args{-having});
    $sql_having =~ s/\bWHERE\b/HAVING/;
    $sql.= " $sql_having";
    push @bind, @bind_having;
  }

  # add ORDER BY if needed
  if (my $order = $args{-order_by}) {

    my ($sql_order, @orderby_bind) = $self->_order_by($order);
    $sql .= $sql_order;
    push @bind, @orderby_bind;
  }

  # add LIMIT/OFFSET if needed
  if (defined $args{-limit}) {
    my ($limit_sql, @limit_bind)
      = $self->limit_offset(@args{qw/-limit -offset/});
    $sql = $limit_sql =~ /%s/ ? sprintf $limit_sql, $sql
                              : "$sql $limit_sql";
    push @bind, @limit_bind;
  }

  # add FOR clause if needed
  my $for = exists $args{-for} ? $args{-for} : $self->{select_implicitly_for};
  $sql .= " FOR $for" if $for;

  # initial WITH clause
  $self->_prepend_WITH_clause(\$sql, \@bind);

  # return results
  if ($args{-want_details}) {
    return {sql             => $sql,
            bind            => \@bind,
            aliased_tables  => ($join_info && $join_info->{aliased_tables}),
            aliased_columns => \%aliased_columns          };
  }
  else {
    return ($sql, @bind);
  }
}

#----------------------------------------------------------------------
# insert
#----------------------------------------------------------------------

sub _setup_insert_inheritance {
  my ($parent_sqla) = @_;

  # if the parent has method '_expand_insert_value' (SQL::Abstract >= v2.0),
  # we need to override it in this subclass
  if ($parent_sqla->can('_expand_insert_value')) {
    *_expand_insert_value = sub {
      my ($self, $v) = @_;

      my $k = our $Cur_Col_Meta;

      if (ref($v) eq 'ARRAY') {
        if ($self->{array_datatypes} || $self->is_bind_value_with_type($v)) {
          return +{ -bind => [ $k, $v ] };
        }
        my ($sql, @bind) = @$v;
        $self->_assert_bindval_matches_bindtype(@bind);
        return +{ -literal => $v };
      }
      if (ref($v) eq 'HASH') {
        if (grep !/^-/, keys %$v) {
          belch "HASH ref as bind value in insert is not supported";
          return +{ -bind => [ $k, $v ] };
        }
      }
      if (!defined($v)) {
        return +{ -bind => [ $k, undef ] };
      }
      return $self->expand_expr($v);
    };
  }

  # otherwise, if the parent is an old SQL::Abstract or it is SQL::Abstract::Classic
  elsif ($parent_sqla->can('_insert_values')) {

    # if the parent has no method '_insert_value', this is the old
    # monolithic _insert_values() method. We must override it
    if (!$parent_sqla->can('_insert_value')) {
      *_insert_values = sub {
         my ($self, $data) = @_;

         my (@values, @all_bind);
         foreach my $column (sort keys %$data) {
           my ($values, @bind) = $self->_insert_value($column, $data->{$column});
           push @values, $values;
           push @all_bind, @bind;
         }
         my $sql = $self->_sqlcase('values')." ( ".join(", ", @values)." )";
         return ($sql, @all_bind);
      };
    }

    # now override the _insert_value() method
    *_insert_value = sub {

      # unfortunately, we can't just override the ARRAYREF part, so the whole
      # parent method is copied here

      my ($self, $column, $v) = @_;

      my (@values, @all_bind);
      $self->_SWITCH_refkind($v, {

        ARRAYREF => sub {
          if ($self->{array_datatypes} # if array datatype are activated
                || $self->is_bind_value_with_type($v)) { # or if this is a bind val
            push @values, '?';
            push @all_bind, $self->_bindtype($column, $v);
          }
          else {                  # else literal SQL with bind
            my ($sql, @bind) = @$v;
            $self->_assert_bindval_matches_bindtype(@bind);
            push @values, $sql;
            push @all_bind, @bind;
          }
        },

        ARRAYREFREF => sub {        # literal SQL with bind
          my ($sql, @bind) = @${$v};
          $self->_assert_bindval_matches_bindtype(@bind);
          push @values, $sql;
          push @all_bind, @bind;
        },

        # THINK : anything useful to do with a HASHREF ?
        HASHREF => sub {       # (nothing, but old SQLA passed it through)
          #TODO in SQLA >= 2.0 it will die instead
          belch "HASH ref as bind value in insert is not supported";
          push @values, '?';
          push @all_bind, $self->_bindtype($column, $v);
        },

        SCALARREF => sub {          # literal SQL without bind
          push @values, $$v;
        },

        SCALAR_or_UNDEF => sub {
          push @values, '?';
          push @all_bind, $self->_bindtype($column, $v);
        },

      });

      my $sql = CORE::join(", ", @values);
      return ($sql, @all_bind);
    }
  }
}



sub insert {
  my $self = shift;

  my @old_API_args;
  my $returning_into;
  my $sql_to_add;
  my $fix_RT134127;

  if (&_called_with_named_args) {
    # extract named args and translate to old SQLA API
    my %args = validate(@_, \%params_for_insert);
    $old_API_args[0] = $args{-into}
      or puke "insert(..) : need -into arg";

    if ($args{-values}) {

      # check mutually exclusive parameters
      !$args{$_}
        or puke "insert(-into => .., -values => ...) : cannot use $_ => "
        for qw/-select -columns/;

      $old_API_args[1] = $args{-values};
    }
    elsif ($args{-select}) {
      local $self->{WITH}; # temporarily disable the WITH part during the subquery
      my ($sql, @bind) = $self->select(%{$args{-select}});
      $old_API_args[1] = \ [$sql, @bind];
      if (my $cols = $args{-columns}) {
        $old_API_args[0] .= "(" . CORE::join(", ", @$cols) . ")";
      }
      $fix_RT134127 = 1 if ($SQL::Abstract::VERSION || 0) >= 2.0;
    }
    else {
      puke "insert(-into => ..) : need either -values arg or -select arg";
    }

    # deal with -returning arg
    ($returning_into, my $old_API_options)
      = $self->_compute_returning($args{-returning});
    push @old_API_args, $old_API_options if $old_API_options;

    # SQL to add after the INSERT keyword
    $sql_to_add = $args{-add_sql};
  }
  else {
    @old_API_args = @_;
  }

  # get results from parent method
  my ($sql, @bind) = $self->next::method(@old_API_args);

  # temporary fix for RT#134127 due to a change of behaviour of insert() in SQLA V2.0
  # .. waiting for SQLA to fix RT#134128
  $sql =~ s/VALUES SELECT/SELECT/ if $fix_RT134127;

  # inject more stuff if using Oracle's "RETURNING ... INTO ..."
  if ($returning_into) {
    $sql .= ' INTO ' . join(", ", ("?") x @$returning_into);
    push @bind, @$returning_into;
  }

  # SQL to add after the INSERT keyword
  $sql =~ s/\b(INSERT)\b/$1 $sql_to_add/i if $sql_to_add;

  # initial WITH clause
  $self->_prepend_WITH_clause(\$sql, \@bind);

  return ($sql, @bind);
}

#----------------------------------------------------------------------
# update
#----------------------------------------------------------------------


sub _setup_update_inheritance {
  my ($parent_sqla) = @_;

  # if the parent has method '_expand_update_set_value' (SQL::Abstract >= v2.0),
  # we need to override it in this subclass
  if ($parent_sqla->can('_expand_update_set_values')) {
    *_parent_update            = $parent_sqla->can('update');
    *_expand_update_set_values = sub {
      my ($self, undef, $data) = @_;
      $self->expand_expr({ -list => [
        map {
          my ($k, $set) = @$_;
          $set = { -bind => $_ } unless defined $set;
          +{ -op => [ '=', { -ident => $k }, $set ] };
        }
        map {
          my $k = $_;
          my $v = $data->{$k};
          (ref($v) eq 'ARRAY'
            ? ($self->{array_datatypes} || $self->is_bind_value_with_type($v)
                ? [ $k, +{ -bind => [ $k, $v ] } ]
                : [ $k, +{ -literal => $v } ])
            : do {
                local our $Cur_Col_Meta = $k;
                [ $k, $self->_expand_expr($v) ]
              }
          );
        } sort keys %$data
      ] });
    };
  }


  # otherwise, if the parent is an old SQL::Abstract or it is SQL::Abstract::Classic
  else {
    # if the parent has method '_update_set_values()', it is a SQLA version >=1.85.
    # We can just use its update() method as _parent_update().
    if ($parent_sqla->can('_update_set_values')) {
      *_parent_update = $parent_sqla->can('update');
    }

    # otherwise, it's the old monolithic update() method. We need to supply our own
    # version as _parent_update().
    else {
      *_parent_update = sub {
         my $self    = shift;
         my $table   = $self->_table(shift);
         my $data    = shift || return;
         my $where   = shift;
         my $options = shift;

         # first build the 'SET' part of the sql statement
         puke "Unsupported data type specified to \$sql->update"
           unless ref $data eq 'HASH';

         my ($sql, @all_bind) = $self->_update_set_values($data);
         $sql = $self->_sqlcase('update ') . $table . $self->_sqlcase(' set ')
                 . $sql;

         if ($where) {
           my($where_sql, @where_bind) = $self->where($where);
           $sql .= $where_sql;
           push @all_bind, @where_bind;
         }

         if ($options->{returning}) {
           my ($returning_sql, @returning_bind) = $self->_update_returning($options);
           $sql .= $returning_sql;
           push @all_bind, @returning_bind;
         }

         return wantarray ? ($sql, @all_bind) : $sql;
       };
      *_update_returning = sub {
         my ($self, $options) = @_;

         my $f = $options->{returning};

         my $fieldlist = $self->_SWITCH_refkind($f, {
           ARRAYREF     => sub {join ', ', map { $self->_quote($_) } @$f;},
           SCALAR       => sub {$self->_quote($f)},
           SCALARREF    => sub {$$f},
         });
         return $self->_sqlcase(' returning ') . $fieldlist;
      };
    }

    # now override or supply the _update_set_value() method
    *_update_set_values = sub {
      my ($self, $data) = @_;

      my (@set, @all_bind);
      for my $k (sort keys %$data) {
        my $v = $data->{$k};
        my $r = ref $v;
        my $label = $self->_quote($k);

        $self->_SWITCH_refkind($v, {
          ARRAYREF => sub {
            if ($self->{array_datatypes}                  # array datatype
                || $self->is_bind_value_with_type($v)) {  # or bind value with type
              push @set, "$label = ?";
              push @all_bind, $self->_bindtype($k, $v);
            }
            else {                          # literal SQL with bind
              my ($sql, @bind) = @$v;
              $self->_assert_bindval_matches_bindtype(@bind);
              push @set, "$label = $sql";
              push @all_bind, @bind;
            }
          },
          ARRAYREFREF => sub { # literal SQL with bind
            my ($sql, @bind) = @${$v};
            $self->_assert_bindval_matches_bindtype(@bind);
            push @set, "$label = $sql";
            push @all_bind, @bind;
          },
          SCALARREF => sub {  # literal SQL without bind
            push @set, "$label = $$v";
          },
          HASHREF => sub {
            my ($op, $arg, @rest) = %$v;

            puke 'Operator calls in update must be in the form { -op => $arg }'
              if (@rest or not $op =~ /^\-(.+)/);

            local $self->{_nested_func_lhs} = $k;
            my ($sql, @bind) = $self->_where_unary_op($1, $arg);

            push @set, "$label = $sql";
            push @all_bind, @bind;
          },
          SCALAR_or_UNDEF => sub {
            push @set, "$label = ?";
            push @all_bind, $self->_bindtype($k, $v);
          },
        });
      }
      # generate sql
      my $sql = CORE::join ', ', @set;
      return ($sql, @all_bind);
    };
  }
}

sub update {
  my $self = shift;

  my $join_info;
  my @old_API_args;
  my $returning_into;
  my %args;
  if (&_called_with_named_args) {
    %args = validate(@_, \%params_for_update);

    # compute join info if the datasource is a join
    $join_info = $self->_compute_join_info($args{-table});
    $args{-table} = \($join_info->{sql}) if $join_info;

    @old_API_args = @args{qw/-table -set -where/};

    # deal with -returning arg
    ($returning_into, my $old_API_options)
      = $self->_compute_returning($args{-returning});
    push @old_API_args, $old_API_options if $old_API_options;
  }
  else {
    @old_API_args = @_;
  }

  # call parent method and merge with bind values from $join_info
  my ($sql, @bind) = $self->_parent_update(@old_API_args);

  unshift @bind, @{$join_info->{bind}} if $join_info;

  # handle additional args if needed
  $self->_handle_additional_args_for_update_delete(\%args, \$sql, \@bind, qr/UPDATE/);

  # inject more stuff if using Oracle's "RETURNING ... INTO ..."
  if ($returning_into) {
    $sql .= ' INTO ' . join(", ", ("?") x @$returning_into);
    push @bind, @$returning_into;
  }

  # initial WITH clause
  $self->_prepend_WITH_clause(\$sql, \@bind);

  return ($sql, @bind);
}






#----------------------------------------------------------------------
# delete
#----------------------------------------------------------------------

sub delete {
  my $self = shift;

  my @old_API_args;
  my %args;
  if (&_called_with_named_args) {
    %args = validate(@_, \%params_for_delete);
    @old_API_args = @args{qw/-from -where/};
  }
  else {
    @old_API_args = @_;
  }

  # call parent method
  my ($sql, @bind) = $self->next::method(@old_API_args);

  # maybe need to handle additional args
  $self->_handle_additional_args_for_update_delete(\%args, \$sql, \@bind, qr/DELETE/);

  # initial WITH clause
  $self->_prepend_WITH_clause(\$sql, \@bind);

  return ($sql, @bind);
}




#----------------------------------------------------------------------
# auxiliary methods for insert(), update() and delete()
#----------------------------------------------------------------------

sub _compute_returning {
  my ($self, $arg_returning) = @_;

  my ($returning_into, $old_API_options);

  if ($arg_returning) {
    # if present, "-returning" may be a scalar, arrayref or hashref; the latter
    # is interpreted as .. RETURNING ... INTO ...


    if (does $arg_returning, 'HASH') {
      my @keys = sort keys %$arg_returning
        or puke "-returning => {} : the hash is empty";

      $old_API_options = {returning => \@keys};
      $returning_into  = [@{$arg_returning}{@keys}];
    }
    else {
      $old_API_options = {returning => $arg_returning};
    }
  }

  return ($returning_into, $old_API_options);
}


sub _handle_additional_args_for_update_delete {
  my ($self, $args, $sql_ref, $bind_ref, $keyword_regex) = @_;

  if (defined $args->{-order_by}) {
    my ($sql_ob, @bind_ob) = $self->_order_by($args->{-order_by});
    $$sql_ref .= $sql_ob;
    push @$bind_ref, @bind_ob;
  }
  if (defined $args->{-limit}) {
    # can't call $self->limit_offset(..) because there shouldn't be any offset
    $$sql_ref .= $self->_sqlcase(' limit ?');
    push @$bind_ref, $args->{-limit};
  }
  if (defined $args->{-add_sql}) {
    $$sql_ref =~ s/\b($keyword_regex)\b/$1 $args->{-add_sql}/i;
  }
}


sub _order_by {
  my ($self, $order) = @_;

  # force scalar into an arrayref
  $order = [$order] if not ref $order;

  # restructure array data
  if (does $order, 'ARRAY') {
    my @clone = @$order;      # because we will modify items

    # '-' and '+' prefixes are translated into {-desc/asc => } hashrefs
    foreach my $item (@clone) {
      next if !$item or ref $item;
      $item =~ s/^-//  and $item = {-desc => $item} and next;
      $item =~ s/^\+// and $item = {-asc  => $item};
    }
    $order = \@clone;
  }

  return $self->next::method($order);
}

#----------------------------------------------------------------------
# other public methods
#----------------------------------------------------------------------

# same pattern for 3 invocation methods
foreach my $attr (qw/table_alias column_alias limit_offset/) {
  no strict 'refs';
  *{$attr} = sub {
    my $self = shift;
    my $method = $self->{$attr}; # grab reference to method body
    $self->$method(@_);          # invoke
  };
}

# readonly accessor methods
foreach my $key (qw/join_syntax  join_assoc_right
                    max_members_IN multicols_sep  has_multicols_in_SQL/) {
  no strict 'refs';
  *{$key} = sub {shift->{$key}};
}


# invocation method for 'join'
sub join {
  my $self = shift;

  # start from the right if right-associative
  @_ = reverse @_ if $self->{join_assoc_right};

  # shift first single item (a table) before reducing pairs (op, table)
  my $combined = shift;
  $combined    = $self->_parse_table($combined)      unless ref $combined;

  # reduce pairs (op, table)
  while (@_) {
    # shift 2 items : next join specification and next table
    my $join_spec  = shift;
    my $table_spec = shift or puke "improper number of operands";

    $join_spec  = $self->_parse_join_spec($join_spec) unless ref $join_spec;
    $table_spec = $self->_parse_table($table_spec)    unless ref $table_spec;
    $combined   = $self->_single_join($combined, $join_spec, $table_spec);
  }

  return $combined; # {sql=> .., bind => [..], aliased_tables => {..}}
}


# utility for merging several "where" clauses
sub merge_conditions {
  my $self = shift;
  my %merged;

  foreach my $cond (@_) {
    if    (does $cond, 'HASH')  {
      foreach my $col (sort keys %$cond) {
        $merged{$col} = $merged{$col} ? [-and => $merged{$col}, $cond->{$col}]
                                      : $cond->{$col};
      }
    }
    elsif (does $cond, 'ARRAY') {
      $merged{-nest} = $merged{-nest} ? {-and => [$merged{-nest}, $cond]}
                                      : $cond;
    }
    elsif ($cond) {
      $merged{$cond} = \"";
    }
  }
  return \%merged;
}

# utility for calling either bind_param or bind_param_inout
our $INOUT_MAX_LEN = 99; # chosen arbitrarily; see L
sub bind_params {
  my ($self, $sth, @bind) = @_;
  $sth->isa('DBI::st') or puke "sth argument is not a DBI statement handle";
  foreach my $i (0 .. $#bind) {
    my $val = $bind[$i];
    if (does $val, 'SCALAR') {
      # a scalarref is interpreted as an INOUT parameter
      $sth->bind_param_inout($i+1, $val, $INOUT_MAX_LEN);
    }
    elsif (does $val, 'ARRAY' and
             my ($bind_meth, @args) = $self->is_bind_value_with_type($val)) {
      # either 'bind_param' or 'bind_param_inout', with 2 or 3 args
      $sth->$bind_meth($i+1, @args);
    }
    else {
      # other cases are passed directly to DBI::bind_param
      $sth->bind_param($i+1, $val);
    }
  }
}

sub is_bind_value_with_type {
  my ($self, $val) = @_;

  # compatibility with DBIx::Class syntax of shape [\%args => $val],
  # see L
  if (   @$val == 2
      && does($val->[0], 'HASH')
      && grep {$val->[0]{$_}} qw/dbd_attrs sqlt_size
                                 sqlt_datatype dbic_colname/) {
    my $args = $val->[0];
    if (my $attrs = $args->{dbd_attrs}) {
      return (bind_param => $val->[1], $attrs);
    }
    elsif (my $size = $args->{sqlt_size}) {
      return (bind_param_inout => $val, $size);
    }
    # other options like 'sqlt_datatype', 'dbic_colname' are not supported
    else {
      puke "unsupported options for bind type : "
           . CORE::join(", ", sort keys %$args);
    }

    # NOTE : the following DBIx::Class shortcuts are not supported
    #  [ $name => $val ] === [ { dbic_colname => $name }, $val ]
    #  [ \$dt  => $val ] === [ { sqlt_datatype => $dt }, $val ]
    #  [ undef,   $val ] === [ {}, $val ]
  }

  # in all other cases, this is not a bind value with type
  return ();
}

#----------------------------------------------------------------------
# private utility methods for 'join'
#----------------------------------------------------------------------

sub _compute_join_info {
  my ($self, $table_arg) = @_;

  if (does($table_arg, 'ARRAY') && $table_arg->[0] eq '-join') {
    my @join_args = @$table_arg;
    shift @join_args;                # drop initial '-join'
    return $self->join(@join_args);
  }
  else {
    return;
  }
}

sub _parse_table {
  my ($self, $table) = @_;

  # extract alias, if any (recognized as "table|alias")
  ($table, my $alias) = split /\|/, $table, 2;

  # build a table spec
  return {
    sql            => $self->table_alias($table, $alias),
    bind           => [],
    name           => ($alias || $table),
    aliased_tables => {$alias ? ($alias => $table) : ()},
   };
}

sub _parse_join_spec {
  my ($self, $join_spec) = @_;

  # parse the join specification
  $join_spec
    or puke "empty join specification";
  my ($op, $bracket, $cond_list) = ($join_spec =~ $self->{join_regex})
    or puke "incorrect join specification : $join_spec\n$self->{join_regex}";
  $op        ||= '<=>';
  $bracket   ||= '{';
  $cond_list ||= '';

  # extract constants (strings between quotes), replaced by placeholders
  my $regex = qr/'       # initial quote
                 (       # begin capturing group
                  [^']*    # any non-quote chars
                  (?:        # begin non-capturing group
                     ''        # pair of quotes
                     [^']*     # any non-quote chars
                  )*         # this non-capturing group 0 or more times
                 )       # end of capturing group
                 '       # ending quote
                /x;
  my $placeholder = '_?_'; # unlikely to be counfounded with any value
  my @constants;
  while ($cond_list =~ s/$regex/$placeholder/) {
    push @constants, $1;
  };
  s/''/'/g for @constants;  # replace pairs of quotes by single quotes

  # accumulate conditions as pairs ($left => \"$op $right")
  my @conditions;
  my @using;
  foreach my $cond (split /,\s*/, $cond_list) {
    # parse the condition (left and right operands + comparison operator)
    my ($left, $cmp, $right) = split /([<>=!^]{1,2})/, $cond;
    if ($cmp && $right) {
      # if operands are not qualified by table/alias name, add sprintf hooks
      $left  = '%1$s.' . $left   unless $left  =~ /\./;
      $right = '%2$s.' . $right  unless $right =~ /\./ or $right eq $placeholder;

      # add this pair into the list; right operand is either a bind value
      # or an identifier within the right table
      $right = $right eq $placeholder ? shift @constants : {-ident => $right};
      push @conditions, $left, {$cmp => $right};
    }
    elsif ($cond =~ /^\w+$/) {
      push @using, $cond;
    }
    else {puke "can't parse join condition: $cond"}
  }

  # build join hashref
  my $join_hash = {operator  => $op};
  $join_hash->{using} = \@using                        if @using;
  $join_hash->{condition}
    = $bracket eq '[' ? [@conditions] : {@conditions}  if @conditions;

  return $join_hash;
}

sub _single_join {
  my $self = shift;

  # if right-associative, restore proper left-right order in pair
  @_ = reverse @_ if $self->{join_assoc_right};
  my ($left, $join_spec, $right) = @_;

  # syntax for assembling all elements
  my $syntax = $self->{join_syntax}{$join_spec->{operator}};

  my ($sql, @bind);

  { no if $] ge '5.022000', warnings => 'redundant';
    # because sprintf instructions  may _intentionally_ omit %.. parameters

    if ($join_spec->{using}) {
      not $join_spec->{condition}
        or puke "join specification has both {condition} and {using} fields";

      $syntax =~ s/\bON\s+%s/USING (%s)/;
      $sql = CORE::join ",", @{$join_spec->{using}};
    }
    elsif ($join_spec->{condition}) {
      not $join_spec->{using}
        or puke "join specification has both {condition} and {using} fields";

      # compute the "ON" clause
      ($sql, @bind) = $self->where($join_spec->{condition});
      $sql =~ s/^\s*WHERE\s+//;

      # substitute left/right tables names for '%1$s', '%2$s'
      $sql = sprintf $sql, $left->{name}, $right->{name};
    }

    # build the final sql
    $sql = sprintf $syntax, $left->{sql}, $right->{sql}, $sql;
  }

  # add left/right bind parameters (if any) into the list
  unshift @bind, @{$left->{bind}}, @{$right->{bind}};

  # build result and return
  my %result = (sql => $sql, bind => \@bind);
  $result{name} = ($self->{join_assoc_right} ? $left : $right)->{name};
  $result{aliased_tables} = $left->{aliased_tables};
  foreach my $alias (keys %{$right->{aliased_tables}}) {
    $result{aliased_tables}{$alias} = $right->{aliased_tables}{$alias};
  }

  return \%result;
}


#----------------------------------------------------------------------
# override of parent's "_where_field_IN"
#----------------------------------------------------------------------

sub _where_field_IN {
  my ($self, $k, $op, $vals) = @_;

  # special algorithm if the key is multi-columns (contains a multicols_sep)
  if ($self->{multicols_sep}) {
    my @cols = split m[$self->{multicols_sep}], $k;
    if (@cols > 1) {
      if ($self->{has_multicols_in_SQL}) {
        # DBMS accepts special SQL syntax for multicolumns
        return $self->_multicols_IN_through_SQL(\@cols, $op, $vals);
      }
      else {
        # DBMS doesn't accept special syntax, so we must use boolean logic
        return $self->_multicols_IN_through_boolean(\@cols, $op, $vals);
      }
    }
  }

  # special algorithm if the number of values exceeds the allowed maximum
  my $max_members_IN = $self->{max_members_IN};
  if ($max_members_IN && does($vals, 'ARRAY')
                      &&  @$vals > $max_members_IN) {
    my @vals = @$vals;
    my @slices;
    while (my @slice = splice(@vals, 0, $max_members_IN)) {
      push @slices, \@slice;
    }
    my @clauses = map {{-$op, $_}} @slices;
    my $connector = $op =~ /^not/i ? '-and' : '-or';
    unshift @clauses, $connector;
    my ($sql, @bind) = $self->where({$k => \@clauses});
    $sql =~ s/\s*where\s*\((.*)\)/$1/i;
    return ($sql, @bind);
  }


  # otherwise, call parent method
  $vals = [@$vals] if blessed $vals; # because SQLA dies on blessed arrayrefs
  return $self->next::method($k, $op, $vals);
}


sub _multicols_IN_through_SQL {
  my ($self, $cols, $op, $vals) = @_;

  # build initial sql
  my $n_cols   = @$cols;
  my $sql_cols = CORE::join(',', map {$self->_quote($_)} @$cols);
  my $sql      = "($sql_cols) " . $self->_sqlcase($op);

  # dispatch according to structure of $vals
  return $self->_SWITCH_refkind($vals, {

    ARRAYREF => sub {    # list of tuples
      # deal with special case of empty list (like the parent class)
      my $n_tuples = @$vals;
      if (!$n_tuples) {
        my $sql = ($op =~ /\bnot\b/i) ? $self->{sqltrue} : $self->{sqlfalse};
        return ($sql);
      }

      # otherwise, build SQL and bind values for the list of tuples
      my @bind;
      foreach my $val (@$vals) {
        does($val, 'ARRAY')
          or $val = [split  m[$self->{multicols_sep}], $val];
        @$val == $n_cols
          or puke "op '$op' with multicols: tuple with improper number of cols";
        push @bind, @$val;
      }
      my $single_tuple = "(" . CORE::join(',', (('?') x $n_cols)) . ")";

      my $all_tuples   = CORE::join(', ', (($single_tuple) x $n_tuples));
      $sql            .= " ($all_tuples)";
      return ($sql, @bind);
    },

    SCALARREF => sub {   # literal SQL
      $sql .= " ($$vals)";
      return ($sql);
    },

    ARRAYREFREF => sub { # literal SQL with bind
      my ($inner_sql, @bind) = @$$vals;
      $sql .= " ($inner_sql)";
      return ($sql, @bind);
    },

    FALLBACK => sub {
      puke "op '$op' with multicols requires a list of tuples or literal SQL";
    },

   });
}


sub _multicols_IN_through_boolean {
  my ($self, $cols, $op, $vals) = @_;

  # can't handle anything else than a list of tuples
  does($vals, 'ARRAY') && @$vals
    or puke "op '$op' with multicols requires a non-empty list of tuples";

  # assemble SQL
  my $n_cols   = @$cols;
  my $sql_cols = CORE::join(' AND ', map {$self->_quote($_) . " = ?"} @$cols);
  my $sql      = "(" . CORE::join(' OR ', (("($sql_cols)") x @$vals)) . ")";
  $sql         = "NOT $sql" if $op =~ /\bnot\b/i;

  # assemble bind values
  my @bind;
  foreach my $val (@$vals) {
    does($val, 'ARRAY')
      or $val = [split  m[$self->{multicols_sep}], $val];
    @$val == $n_cols
      or puke "op '$op' with multicols: tuple with improper number of cols";
    push @bind, @$val;
  }

  # return the whole thing
  return ($sql, @bind);
}



#----------------------------------------------------------------------
# override of parent's methods for decoding arrayrefs
#----------------------------------------------------------------------

sub _where_hashpair_ARRAYREF {
  my ($self, $k, $v) = @_;

  if ($self->is_bind_value_with_type($v)) {
    $self->_assert_no_bindtype_columns;
    my $sql = CORE::join ' ', $self->_convert($self->_quote($k)),
                              $self->_sqlcase($self->{cmp}),
                              $self->_convert('?');
    my @bind = ($v);
    return ($sql, @bind);
  }
  else {
    return $self->next::method($k, $v);
  }
}


sub _where_field_op_ARRAYREF {
  my ($self, $k, $op, $vals) = @_;

  if ($self->is_bind_value_with_type($vals)) {
    $self->_assert_no_bindtype_columns;
    my $sql = CORE::join ' ', $self->_convert($self->_quote($k)),
                              $self->_sqlcase($op),
                              $self->_convert('?');
    my @bind = ($vals);
    return ($sql, @bind);
  }
  else {
    return $self->next::method($k, $op, $vals);
  }
}

sub _assert_no_bindtype_columns {
  my ($self) = @_;
  $self->{bindtype} ne 'columns'
    or puke 'values of shape [$val, \%type] are not compatible'
          . 'with ...->new(bindtype => "columns")';
}



#----------------------------------------------------------------------
# method creations through closures
#----------------------------------------------------------------------

sub _make_sub_column_alias {
  my ($self) = @_;
  my $syntax = $self->{column_alias};
  $self->{column_alias} = sub {
    my ($self, $name, $alias) = @_;
    return $name if !$alias;

    # quote $name unless it is an SQL expression (then the user should quote it)
    $name = $self->_quote($name) unless $name =~ /[()]/;

    # assemble syntax
    my $sql = sprintf $syntax, $name, $self->_quote($alias);

    # return a string ref to avoid quoting by SQLA
    return \$sql;
  };
}


sub _make_sub_table_alias {
  my ($self) = @_;
  my $syntax = $self->{table_alias};
  $self->{table_alias} = sub {
    my ($self, $name, $alias) = @_;
    return $name if !$alias;

    # assemble syntax
    my $sql = sprintf $syntax, $self->_quote($name), $self->_quote($alias);

    return $sql;
  };
}



sub _choose_LIMIT_OFFSET_dialect {
  my $self = shift;
  my $dialect = $self->{limit_offset};
  my $method = $limit_offset_dialects{$dialect}
    or puke "no such limit_offset dialect: $dialect";
  $self->{limit_offset} = $method;
}


#----------------------------------------------------------------------
# utility to decide if the method was called with named or positional args
#----------------------------------------------------------------------

sub _called_with_named_args {
  return $_[0] && !ref $_[0]  && substr($_[0], 0, 1) eq '-';
}


1; # End of SQL::Abstract::More

__END__