File Coverage

blib/lib/SQL/Abstract/More.pm

Criterion	Covered	Total	%
statement	547	645	84.8
branch	227	316	71.8
condition	57	93	61.2
subroutine	69	82	84.1
pod	16	16	100.0
total	916	1152	79.5

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

# no "use parent ..." here -- for the moment the inheritance is specified dynamically in the
# import() method -- inheriting either from SQL::Abstract or SQL::Abstract::Classic.
# This is deprecated and will no longer be supported in the next version

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.49_01';
our @ISA;

sub import {
  my $class = shift;

  # the parent class may be specified from an environment variable ...
  my $parent_sqla = $ENV{SQL_ABSTRACT_MORE_EXTENDS};

  # ... or from param -extends => .. when calling import() ...
  $parent_sqla = $_[1] if @_ >= 2 && $_[0] eq '-extends';

  # ... but choosing the parent is deprecated!
  ! $parent_sqla
    or warn "explicitly specification of $parent_sqla as parent class to SQL::Abstract::More is deprecated; "
          . "future versions will no longer offer that possibility";

  # default parent class .. this will be mandatory in the next version. Later versions may have no parent at all.
  $parent_sqla ||= 'SQL::Abstract::Classic';

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

  # make sure that import() is never 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;},
  OffsetFetchRows => sub {my ($self, $limit, $offset) = @_;
                          $offset ||= 0;
                          return "OFFSET ? ROWS FETCH NEXT ? ROWS ONLY", $offset, $limit;},
  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 (for Oracle < 12c).
    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,                       },
);
$sql_dialects{Oracle12c} = {%{$sql_dialects{Oracle}}, limit_offset => "OffsetFetchRows"};


# 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},
  -as           => {type => SCALAR,                  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;

  # make sure @ISA is populated, in case import() was not already called
  $class->import() if !@ISA;

  # accept params either as hash or as hashref
  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 ... but explict params above had precedence
  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 additional attributes into $self
  $self->{$_} = $more_self->{$_} foreach keys %$more_self;

  # arguments supplied as scalars are transformed into coderefs
  ref $self->{column_alias} or $self->_make_aliasing_sub('column_alias');
  ref $self->{table_alias}  or $self->_make_aliasing_sub('table_alias');
  ref $self->{limit_offset} or $self->_choose_LIMIT_OFFSET_dialect;

  # compute the regex for parsing join specifications - depending on 'join_syntax' attribute
  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 {
  my $self = shift;

  ! $self->{WITH}
    or puke "calls to methods with() or with_recursive() 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 @cols = does($args{-columns}, 'ARRAY') ? @{$args{-columns}}
             : $args{-columns}                ? ($args{-columns})
             :                                  ();
    my ($sql, @bind) = $self->select(%{$args{-as_select}});
    $clone->{WITH}{sql} .= ", " if $clone->{WITH}{sql};
    $clone->{WITH}{sql} .= $self->_quote($args{-table});
    $clone->{WITH}{sql} .= "(" . join(", ", map {$self->_quote($_)} @cols) . ")"   if @cols;
    $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 with_recursive {
  my $self = shift;

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

  return $new_instance;
}

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 this method was called with positional args, just delegate to the parent
  return $self->next::method(@_) if !&_called_with_named_args;

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

  # infrastructure for collecting fragments of sql and bind args
  my ($sql, @bind) = ("");
  my $add_sql_bind = sub {$sql .= shift; push @bind, @_}; # closure to add to ($sql, @bind)

  # parse columns and datasource
  my $from                                               = $self->_parse_datasource($args{-from});
  my ($cols, $post_select, $cols_bind, $aliased_columns) = $self->_parse_columns($args{-columns});

  # parse the WHERE conditions
  my ($where_sql, @where_bind) = $self->where($args{-where});

  # assemble the SELECT statement
  my $fields     = ref $cols ? join ", ", @$cols : $cols;
  my $select_sql = join(' ', $self->_sqlcase('select'), $fields, $self->_sqlcase('from'), $from->{sql}) . $where_sql;
  $add_sql_bind->($select_sql, @$cols_bind, @{$from->{bind}}, @where_bind);

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

  # add set operators (UNION, INTERSECT, etc) if needed
  foreach my $set_op (@set_operators) {
    if (my $val_set_op = $args{-$set_op}) {
      my ($sql_set_op, @bind_set_op) = $self->_parse_set_operator($set_op => $val_set_op, $args{-columns}, $args{-from});
      $add_sql_bind->($sql_set_op, @bind_set_op);
    }
  }

  # add GROUP BY if needed
  if ($args{-group_by}) {
    my $sql_grp = $self->where(undef, $args{-group_by});
    $sql_grp =~ s/\bORDER\b/GROUP/i;
    $add_sql_bind->($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/i;
    $add_sql_bind->(" $sql_having", @bind_having);
  }

  # add ORDER BY if needed
  if (my $order = $args{-order_by}) {
    $add_sql_bind->($self->_order_by($order));
  }

  # add pagination if needed (either -page_* args or -limit/-offset)
  $self->_translate_page_into_limit_offset(\%args) if $args{-page_index} or $args{-page_size};
  if (defined $args{-limit}) {
    my ($limit_sql, @limit_bind) = $self->limit_offset(@args{qw/-limit -offset/});
    if ($limit_sql =~ /%s/) {
      $sql = sprintf $limit_sql, $sql; # rewrite the whole $sql
      push @bind, @limit_bind;
    }
    else {
      $add_sql_bind->(" $limit_sql", @limit_bind);
    }
  }

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

  # add alias if select() is used as a subquery
  if (my $alias = $args{-as}) {
    $sql = "($sql)|$alias";
  }

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

  # return results
  return $args{-want_details} ? {aliased_tables  => $from->{aliased_tables},
                                 aliased_columns => $aliased_columns,
                                 sql             => $sql,
                                 bind            => \@bind}
                              : ($sql, @bind);
}


sub _parse_columns {
  my ($self, $columns) = @_;

  # the -columns arg can be an arrayref or a plain scalar => unify into an array
  my @cols = ref $columns ? @$columns : ($columns);

  # initial members of the columns list starting with "-" are extracted into a
  # separate list @post_select, later re-injected into the SQL (for ex. '-distinct')
  my @post_select;
  push @post_select, shift @cols while @cols && $cols[0] =~ s/^-//;

  # loop over columns, handling aliases and subqueries
  my @cols_bind;
  my %aliased_columns;
  foreach my $col (@cols) {
    # deal with subquery of shape \ [$sql, @bind]
    if (_is_subquery($col)) {
      my ($sql, @col_bind) = @$$col;
      $col = _parenthesize_select($sql);
      push @cols_bind, @col_bind;
    }

    # check for a column alias; if present, register it in the alias table
    ($col, my $alias)        = $self->_parse_alias($col);
    $aliased_columns{$alias} = $col if $alias ;

    # quote the column SQL if necessary - excluding expressions with parens or commas
    # (such as SQL functions or parenthesized subqueries)
    $col = $self->_quote($col) unless $col =~ /[(),]/;

    # insert SQL aliasing if necessary
    $col = $self->column_alias($col, $self->_quote($alias)) if $alias;
  }

  return (\@cols, \@post_select, \@cols_bind, \%aliased_columns);
}





sub _parse_set_operator {
  my ($self, $set_op, $val_set_op, $cols, $from) = @_;

  my %sub_args = @$val_set_op;
  $sub_args{-columns} ||= $cols;
  $sub_args{-from}    ||= $from;
  local $self->{WITH}; # temporarily disable the WITH part during the recursive call to select()
  my ($sql, @bind) = $self->select(%sub_args);
  (my $sql_op = uc($set_op)) =~ s/_/ /g; # for ex. -union_all becomes 'UNION ALL'
  return (" $sql_op $sql", @bind);
}


sub _translate_page_into_limit_offset {
  my ($self, $args) = @_;

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



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

    # we also need to back-implement the _insert_values method from earlier versions of SQLA
    *_insert_values = sub {
      my $self = shift;
      my ($sql, @bind) = $self->SUPER::insert(FAKE_TABLE => @_);
      $sql =~ s/^.*\bVALUES\b/VALUES/i;
      return ($sql, @bind);
    };

  }

  # 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);
    }
  }
  else {
    puke "unexpected parent class, cannot setup inheritance rules for insert()";
  }
}



sub insert {
  my $self = shift;

  my ($sql, @bind) = ("");
  my $add_sql_bind = sub {$sql .= shift; push @bind, @_}; # closure for adding to ($sql, @bind)

  if (not &_called_with_named_args) {
    ($sql, @bind) = $self->next::method(@_);
  }
  else {
    # extract named args
    my %args = validate(@_, \%params_for_insert);

    $args{-into} or puke "insert(..) : need -into arg";
    my $source = $self->_parse_datasource($args{-into});
    $add_sql_bind->($self->_sqlcase('insert into') . " " . $source->{sql},
                    @{$source->{bind}}, # in principle always empty for an insert, but just for consistency's sake
                   );

    if (my $values = $args{-values}) {
      !$args{-select}  or puke "insert(-into => .., -values => ...) : incompatible with -select => ";

      if (does($values, 'HASH')) {
        !$args{-columns} or puke "insert(-into => .., -values => {...}) : incompatible with -columns => ";
        my $quoted_cols = join ", ", map {$self->_quote($_)} sort keys %$values;
        $add_sql_bind->("($quoted_cols)");
      }
      elsif (does($values, 'ARRAY')) {
        if (my $cols = $args{-columns}) {
          $cols = [$cols] if ! ref $cols;
          @$cols == @$values or puke "insert(-into => .., -columns => [...], -values => [...]) : numbers of columns do not match";
          my %merged; @merged{@$cols} = @$values;
          $values = \%merged;
          # NOTE: in doing so we lost the order of @$cols. In principle this has no impact on the final
          # result, but nevertheless is not very satisfactory. In a future version when we no longer need
          # to support SQLA 2.0, we should have our own version of $self->_insert_cols_values($cols, $values);

          my $quoted_cols = join ", ", map {$self->_quote($_)} sort @$cols;
          $add_sql_bind->("($quoted_cols)");
        }
        else {
          # Code and comments below borrowed from SQL::Abstract!
          #
          # fold the list of values into a hash of column name - value pairs
          # (where the column names are artificially generated, and their
          # lexicographical ordering keep the ordering of the original list)
          my $i = "a"; # incremented values will be in lexicographical order
          $values = { map { ($i++ => $_) } @$values };
        }
      }
      else {
        puke "insert(-into => .., -values => ...) : -values must be a hashref or an arrayref";
      }

      my ($val_sql, @val_bind)   = $self->_insert_values($values);
      $add_sql_bind->(" $val_sql", @val_bind);
    }
    elsif ($args{-select}) {
      local $self->{WITH}; # temporarily disable the WITH part during the subquery
      if (my $cols = $args{-columns}) {
        $cols = [$cols] if ! ref $cols;
        my $quoted_cols = join ", ", map {$self->_quote($_)} @$cols;
        $add_sql_bind->("($quoted_cols)");
      }
      my ($select_sql, @select_bind) = $self->select(%{$args{-select}});
      $add_sql_bind->(" $select_sql", @select_bind);
    }
    else {
      puke "insert(-into => ..) : need either -values arg or -select arg";
    }

    # THINK: should also support some syntax for generating INSERT INTO Foo(a, b, c) VALUES (1, 2, 3), (4, 5, 6), ...
    # Probably a variant like -columns => [qw/a b c/], -values => [[1, 2, 3], [4, 5, 6], ..]

    # deal with -returning arg
    if ($args{-returning}) {
      my ($into, $returning) = $self->_compute_returning($args{-returning});
      my ($r_sql, @r_bind)   = $self->_insert_returning($returning); # TODO : implement directly instead of calling SQLA
      $add_sql_bind->($r_sql, @r_bind);

      # inject more stuff if using Oracle's "RETURNING ... INTO ..."
      if ($into) {
        $add_sql_bind->($self->_sqlcase(' into ') . join(", ", ("?") x @$into),
                        @$into);
      }
    }

    # SQL to add after the INSERT keyword
    $sql =~ s/\b(INSERT)\b/$1 $args{-add_sql}/i if $args{-add_sql};
  };

  # 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_values() 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 ($sql, @bind);

  if (not &_called_with_named_args) {
    ($sql, @bind) = $self->next::method(@_);
  }
  else {
    my %args = validate(@_, \%params_for_update);

    # first build the 'SET' part of the sql statement
    does($args{-set}, 'HASH') or puke "Unsupported data type specified to \$sql->update";
    ($sql, @bind) = $self->_update_set_values($args{-set});

    # add the initial part of the statement
    my $join_info = $self->_compute_join_info($args{-table});
    my $table     = defined $join_info  ? $join_info->{sql}
                                        : $self->_parse_table($args{-table})->{sql};
    substr($sql, 0, 0) = $self->_sqlcase('update ') . $table . $self->_sqlcase(' set ');

    # add the 'WHERE' part
    if ($args{-where}) {
      my ($where_sql, @where_bind) = $self->where($args{-where});
      $sql .= $where_sql;
      push @bind, @where_bind;
    }

    # deal with -returning arg
    my ($into, $returning) = $self->_compute_returning($args{-returning});
    $sql .= $self->_update_returning($returning) if $returning;

    # merge with bind values from $join_info
    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/);

    # deal with Oracle's "RETURNING ... INTO ..."
    if ($into) {
      $sql .= ' INTO ' . join(", ", ("?") x @$into);
      push @bind, @$into;
    }
  }


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

  return ($sql, @bind);
}






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

sub delete {
  my $self = shift;

  my ($sql, @bind);

  if (not &_called_with_named_args) {
    ($sql, @bind) = $self->next::method(@_);
  }
  else {
    my %args = validate(@_, \%params_for_delete);
    ($sql, @bind)      = $self->where($args{-where});
    substr($sql, 0, 0) = $self->_sqlcase('delete from ') . $self->_parse_table($args{-from})->{sql};

    # handle additional args if needed
    $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
sub table_alias  {my $self = shift; my $meth = $self->{table_alias};  $self->$meth(@_)}
sub column_alias {my $self = shift; my $meth = $self->{column_alias}; $self->$meth(@_)}
sub limit_offset {my $self = shift; my $meth = $self->{limit_offset}; $self->$meth(@_)}



# 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 $first_source = shift;
  my $accumulator  = $self->_parse_datasource($first_source);

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

    $join_spec   = $self->_parse_join_spec($join_spec) unless ref $join_spec;
    $next_source = $self->_parse_datasource($next_source);
    $accumulator = $self->_single_join($accumulator, $join_spec, $next_source);
  }

  return $accumulator; # {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') {
    return $self->join(@{$table_arg}[1 .. $#$table_arg]); # drop initial '-join'
  }
  else {
    return;
  }
}


sub _parse_alias {
  my ($self, $input_sql) = @_;

  return $input_sql =~ /^\s*             # ignore insignificant leading spaces
                         (\S.*?)         # $1: any non-empty string, no starting with space (but may include '|')
                         (?
                         \|              # literal '|' (the last in the string)
                         (\pL[\w\s]*?)   # $2 : alias name: initial letter, then word chars or spaces
                         $/x

  #         sql         alias
  #         ===         =====
         ? ($1,         $2   )
         : ($input_sql, undef);
}


sub _parse_datasource {
  my ($self, $source) = @_;

  if (my $join_info = $self->_compute_join_info($source)) {
    return $join_info;
  }
  elsif (_is_subquery($source)) {
    # separate the $sql and @bind parts
    my ($sql, @bind) = @$$source;

    my $table_spec = $self->_parse_table(_parenthesize_select($sql), -dont_quote_table);
    $table_spec->{bind} = \@bind;
    return $table_spec;
  }
  elsif (does($source, 'ARRAY')) {
    # compatibility with old SQL::Abstract -- deprecated
    my @table_specs = map {$self->_parse_table($_)} @$source;
    return {sql            => (CORE::join ', ', map {$_->{sql}}               @table_specs),
            bind           => [                 map {@{$_->{bind}}}           @table_specs],
            aliased_tables => {                 map {%{$_->{aliased_tables}}} @table_specs}};
  }
  elsif (does($source, 'SCALAR')) {
    # reference to a SQL string --> handled as is without quoting
    return $self->_parse_table($$source, -dont_quote_table);
  }
  else {
    return $self->_parse_table($source);
  }
}



sub _parse_table {
  my ($self, $table, $dont_quote_table) = @_; # $dont_quote_table: just a boolean

  ($table, my $alias) = $self->_parse_alias($table);

  my $table_spec = {
    bind           => [],
    name           => ($alias || $table),
    aliased_tables => {$alias ? ($alias => $table) : ()},
   };

  $table = $self->_quote($table) unless $dont_quote_table;

  $table_spec->{sql} = $alias ? $self->table_alias($table, $self->_quote($alias)) : $table;

  return $table_spec;
}

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 ",", map {$self->_quote($_)} @{$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 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_aliasing_sub {
  my ($self, $syntax_field) = @_;
  my $syntax = $self->{$syntax_field};
  $self->{$syntax_field} = sub {
    my ($self, $name, $alias) = @_;
    return $alias ? sprintf($syntax, $name, $alias) : $name;
  };
}

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 functions (not methods)
#----------------------------------------------------------------------

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


sub _is_subquery {
  my $arg = shift;
  return does($arg, 'REF') && does($$arg, 'ARRAY');
}


sub _parenthesize_select {
  my ($sql) = @_;
  $sql =~ s{^(select\b.*)}{($1)}is; # if subquery is a plain SELECT, put it in parenthesis
  return $sql;
}




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

__END__