lib/DBIx/Class/SQLAHacks.pm

   1 package # Hide from PAUSE
   2   DBIx::Class::SQLAHacks;
   3
   4 # This module is a subclass of SQL::Abstract::Limit and includes a number
   5 # of DBIC-specific workarounds, not yet suitable for inclusion into the
   6 # SQLA core
   7
   8 use base qw/SQL::Abstract::Limit/;
   9 use strict;
  10 use warnings;
  11 use Carp::Clan qw/^DBIx::Class|^SQL::Abstract/;
  12 use Sub::Name();
  13
  14 BEGIN {
  15   # reinstall the carp()/croak() functions imported into SQL::Abstract
  16   # as Carp and Carp::Clan do not like each other much
  17   no warnings qw/redefine/;
  18   no strict qw/refs/;
  19   for my $f (qw/carp croak/) {
  20
  21     my $orig = \&{"SQL::Abstract::$f"};
  22     *{"SQL::Abstract::$f"} = Sub::Name::subname "SQL::Abstract::$f" =>
  23       sub {
  24         if (Carp::longmess() =~ /DBIx::Class::SQLAHacks::[\w]+ .+? called \s at/x) {
  25           __PACKAGE__->can($f)->(@_);
  26         }
  27         else {
  28           goto $orig;
  29         }
  30       };
  31   }
  32 }
  33
  34
  35 # Tries to determine limit dialect.
  36 #
  37 sub new {
  38   my $self = shift->SUPER::new(@_);
  39
  40   # This prevents the caching of $dbh in S::A::L, I believe
  41   # If limit_dialect is a ref (like a $dbh), go ahead and replace
  42   #   it with what it resolves to:
  43   $self->{limit_dialect} = $self->_find_syntax($self->{limit_dialect})
  44     if ref $self->{limit_dialect};
  45
  46   $self;
  47 }
  48
  49
  50 # Slow but ANSI standard Limit/Offset support. DB2 uses this
  51 sub _RowNumberOver {
  52   my ($self, $sql, $order, $rows, $offset ) = @_;
  53
  54   $offset += 1;
  55   my $last = $rows + $offset - 1;
  56   my ( $order_by ) = $self->_order_by( $order );
  57
  58   $sql = <<"SQL";
  59 SELECT * FROM
  60 (
  61    SELECT Q1.*, ROW_NUMBER() OVER( ) AS ROW_NUM FROM (
  62       $sql
  63       $order_by
  64    ) Q1
  65 ) Q2
  66 WHERE ROW_NUM BETWEEN $offset AND $last
  67
  68 SQL
  69
  70   return $sql;
  71 }
  72
  73 # Crappy Top based Limit/Offset support. MSSQL uses this currently,
  74 # but may have to switch to RowNumberOver one day
  75 sub _Top {
  76   my ( $self, $sql, $order, $rows, $offset ) = @_;
  77
  78   # mangle the input sql so it can be properly aliased in the outer queries
  79   $sql =~ s/^ \s* SELECT \s+ (.+?) \s+ (?=FROM)//ix
  80     or croak "Unrecognizable SELECT: $sql";
  81   my $sql_select = $1;
  82   my @sql_select = split (/\s*,\s*/, $sql_select);
  83
  84   # we can't support subqueries (in fact MSSQL can't) - croak
  85   if (@sql_select != @{$self->{_dbic_rs_attrs}{select}}) {
  86     croak (sprintf (
  87       'SQL SELECT did not parse cleanly - retrieved %d comma separated elements, while '
  88     . 'the resultset select attribure contains %d elements: %s',
  89       scalar @sql_select,
  90       scalar @{$self->{_dbic_rs_attrs}{select}},
  91       $sql_select,
  92     ));
  93   }
  94
  95   my $name_sep = $self->name_sep || '.';
  96   my $esc_name_sep = "\Q$name_sep\E";
  97   my $col_re = qr/ ^ (?: (.+) $esc_name_sep )? ([^$esc_name_sep]+) $ /x;
  98
  99   my $rs_alias = $self->{_dbic_rs_attrs}{alias};
 100   my $quoted_rs_alias = $self->_quote ($rs_alias);
 101
 102   # construct the new select lists, rename(alias) some columns if necessary
 103   my (@outer_select, @inner_select, %seen_names, %col_aliases, %outer_col_aliases);
 104
 105   for (@{$self->{_dbic_rs_attrs}{select}}) {
 106     next if ref $_;
 107     my ($table, $orig_colname) = ( $_ =~ $col_re );
 108     next unless $table;
 109     $seen_names{$orig_colname}++;
 110   }
 111
 112   for my $i (0 .. $#sql_select) {
 113
 114     my $colsel_arg = $self->{_dbic_rs_attrs}{select}[$i];
 115     my $colsel_sql = $sql_select[$i];
 116
 117     # this may or may not work (in case of a scalarref or something)
 118     my ($table, $orig_colname) = ( $colsel_arg =~ $col_re );
 119
 120     my $quoted_alias;
 121     # do not attempt to understand non-scalar selects - alias numerically
 122     if (ref $colsel_arg) {
 123       $quoted_alias = $self->_quote ('column_' . (@inner_select + 1) );
 124     }
 125     # column name seen more than once - alias it
 126     elsif ($orig_colname &&
 127           ($seen_names{$orig_colname} && $seen_names{$orig_colname} > 1) ) {
 128       $quoted_alias = $self->_quote ("${table}__${orig_colname}");
 129     }
 130
 131     # we did rename - make a record and adjust
 132     if ($quoted_alias) {
 133       # alias inner
 134       push @inner_select, "$colsel_sql AS $quoted_alias";
 135
 136       # push alias to outer
 137       push @outer_select, $quoted_alias;
 138
 139       # Any aliasing accumulated here will be considered
 140       # both for inner and outer adjustments of ORDER BY
 141       $self->__record_alias (
 142         \%col_aliases,
 143         $quoted_alias,
 144         $colsel_arg,
 145         $table ? $orig_colname : undef,
 146       );
 147     }
 148
 149     # otherwise just leave things intact inside, and use the abbreviated one outside
 150     # (as we do not have table names anymore)
 151     else {
 152       push @inner_select, $colsel_sql;
 153
 154       my $outer_quoted = $self->_quote ($orig_colname);  # it was not a duplicate so should just work
 155       push @outer_select, $outer_quoted;
 156       $self->__record_alias (
 157         \%outer_col_aliases,
 158         $outer_quoted,
 159         $colsel_arg,
 160         $table ? $orig_colname : undef,
 161       );
 162     }
 163   }
 164
 165   my $outer_select = join (', ', @outer_select );
 166   my $inner_select = join (', ', @inner_select );
 167
 168   %outer_col_aliases = (%outer_col_aliases, %col_aliases);
 169
 170   # deal with order
 171   croak '$order supplied to SQLAHacks limit emulators must be a hash'
 172     if (ref $order ne 'HASH');
 173
 174   $order = { %$order }; #copy
 175
 176   my $req_order = $order->{order_by};
 177
 178   # examine normalized version, collapses nesting
 179   my $limit_order;
 180   if (scalar $self->_order_by_chunks ($req_order)) {
 181     $limit_order = $req_order;
 182   }
 183   else {
 184     $limit_order = [ map
 185       { join ('', $rs_alias, $name_sep, $_ ) }
 186       ( $self->{_dbic_rs_attrs}{_source_handle}->resolve->primary_columns )
 187     ];
 188   }
 189
 190   my ( $order_by_inner, $order_by_outer ) = $self->_order_directions($limit_order);
 191   my $order_by_requested = $self->_order_by ($req_order);
 192
 193   # generate the rest
 194   delete $order->{order_by};
 195   my $grpby_having = $self->_order_by ($order);
 196
 197   # short circuit for counts - the ordering complexity is needless
 198   if ($self->{_dbic_rs_attrs}{-for_count_only}) {
 199     return "SELECT TOP $rows $inner_select $sql $grpby_having $order_by_outer";
 200   }
 201
 202   # we can't really adjust the order_by columns, as introspection is lacking
 203   # resort to simple substitution
 204   for my $col (keys %outer_col_aliases) {
 205     for ($order_by_requested, $order_by_outer) {
 206       $_ =~ s/\s+$col\s+/ $outer_col_aliases{$col} /g;
 207     }
 208   }
 209   for my $col (keys %col_aliases) {
 210     $order_by_inner =~ s/\s+$col\s+/ $col_aliases{$col} /g;
 211   }
 212
 213
 214   my $inner_lim = $rows + $offset;
 215
 216   $sql = "SELECT TOP $inner_lim $inner_select $sql $grpby_having $order_by_inner";
 217
 218   if ($offset) {
 219     $sql = <<"SQL";
 220
 221     SELECT TOP $rows $outer_select FROM
 222     (
 223       $sql
 224     ) $quoted_rs_alias
 225     $order_by_outer
 226 SQL
 227
 228   }
 229
 230   if ($order_by_requested) {
 231     $sql = <<"SQL";
 232
 233     SELECT $outer_select FROM
 234       ( $sql ) $quoted_rs_alias
 235     $order_by_requested
 236 SQL
 237
 238   }
 239
 240   $sql =~ s/\s*\n\s*/ /g; # parsing out multiline statements is harder than a single line
 241   return $sql;
 242 }
 243
 244 # action at a distance to shorten Top code above
 245 sub __record_alias {
 246   my ($self, $register, $alias, $fqcol, $col) = @_;
 247
 248   # record qualified name
 249   $register->{$fqcol} = $alias;
 250   $register->{$self->_quote($fqcol)} = $alias;
 251
 252   return unless $col;
 253
 254   # record unqualified name, undef (no adjustment) if a duplicate is found
 255   if (exists $register->{$col}) {
 256     $register->{$col} = undef;
 257   }
 258   else {
 259     $register->{$col} = $alias;
 260   }
 261
 262   $register->{$self->_quote($col)} = $register->{$col};
 263 }
 264
 265
 266
 267 # While we're at it, this should make LIMIT queries more efficient,
 268 #  without digging into things too deeply
 269 sub _find_syntax {
 270   my ($self, $syntax) = @_;
 271   return $self->{_cached_syntax} ||= $self->SUPER::_find_syntax($syntax);
 272 }
 273
 274 my $for_syntax = {
 275   update => 'FOR UPDATE',
 276   shared => 'FOR SHARE',
 277 };
 278 # Quotes table names, handles "limit" dialects (e.g. where rownum between x and
 279 # y), supports SELECT ... FOR UPDATE and SELECT ... FOR SHARE.
 280 sub select {
 281   my ($self, $table, $fields, $where, $order, @rest) = @_;
 282
 283   $self->{"${_}_bind"} = [] for (qw/having from order/);
 284
 285   if (not ref($table) or ref($table) eq 'SCALAR') {
 286     $table = $self->_quote($table);
 287   }
 288
 289   local $self->{rownum_hack_count} = 1
 290     if (defined $rest[0] && $self->{limit_dialect} eq 'RowNum');
 291   @rest = (-1) unless defined $rest[0];
 292   croak "LIMIT 0 Does Not Compute" if $rest[0] == 0;
 293     # and anyway, SQL::Abstract::Limit will cause a barf if we don't first
 294   my ($sql, @where_bind) = $self->SUPER::select(
 295     $table, $self->_recurse_fields($fields), $where, $order, @rest
 296   );
 297   if (my $for = delete $self->{_dbic_rs_attrs}{for}) {
 298     $sql .= " $for_syntax->{$for}" if $for_syntax->{$for};
 299   }
 300
 301   return wantarray ? ($sql, @{$self->{from_bind}}, @where_bind, @{$self->{having_bind}}, @{$self->{order_bind}} ) : $sql;
 302 }
 303
 304 # Quotes table names, and handles default inserts
 305 sub insert {
 306   my $self = shift;
 307   my $table = shift;
 308   $table = $self->_quote($table);
 309
 310   # SQLA will emit INSERT INTO $table ( ) VALUES ( )
 311   # which is sadly understood only by MySQL. Change default behavior here,
 312   # until SQLA2 comes with proper dialect support
 313   if (! $_[0] or (ref $_[0] eq 'HASH' and !keys %{$_[0]} ) ) {
 314     return "INSERT INTO ${table} DEFAULT VALUES"
 315   }
 316
 317   $self->SUPER::insert($table, @_);
 318 }
 319
 320 # Just quotes table names.
 321 sub update {
 322   my $self = shift;
 323   my $table = shift;
 324   $table = $self->_quote($table);
 325   $self->SUPER::update($table, @_);
 326 }
 327
 328 # Just quotes table names.
 329 sub delete {
 330   my $self = shift;
 331   my $table = shift;
 332   $table = $self->_quote($table);
 333   $self->SUPER::delete($table, @_);
 334 }
 335
 336 sub _emulate_limit {
 337   my $self = shift;
 338   if ($_[3] == -1) {
 339     return $_[1].$self->_order_by($_[2]);
 340   } else {
 341     return $self->SUPER::_emulate_limit(@_);
 342   }
 343 }
 344
 345 sub _recurse_fields {
 346   my ($self, $fields, $params) = @_;
 347   my $ref = ref $fields;
 348   return $self->_quote($fields) unless $ref;
 349   return $$fields if $ref eq 'SCALAR';
 350
 351   if ($ref eq 'ARRAY') {
 352     return join(', ', map {
 353       $self->_recurse_fields($_)
 354         .(exists $self->{rownum_hack_count} && !($params && $params->{no_rownum_hack})
 355           ? ' AS col'.$self->{rownum_hack_count}++
 356           : '')
 357       } @$fields);
 358   }
 359   elsif ($ref eq 'HASH') {
 360     my %hash = %$fields;
 361
 362     my $as = delete $hash{-as};   # if supplied
 363
 364     my ($func, $args) = each %hash;
 365     delete $hash{$func};
 366
 367     if (lc ($func) eq 'distinct' && ref $args eq 'ARRAY' && @$args > 1) {
 368       croak (
 369         'The select => { distinct => ... } syntax is not supported for multiple columns.'
 370        .' Instead please use { group_by => [ qw/' . (join ' ', @$args) . '/ ] }'
 371        .' or { select => [ qw/' . (join ' ', @$args) . '/ ], distinct => 1 }'
 372       );
 373     }
 374
 375     my $select = sprintf ('%s( %s )%s',
 376       $self->_sqlcase($func),
 377       $self->_recurse_fields($args),
 378       $as
 379         ? sprintf (' %s %s', $self->_sqlcase('as'), $as)
 380         : ''
 381     );
 382
 383     # there should be nothing left
 384     if (keys %hash) {
 385       croak "Malformed select argument - too many keys in hash: " . join (',', keys %$fields );
 386     }
 387
 388     return $select;
 389   }
 390   # Is the second check absolutely necessary?
 391   elsif ( $ref eq 'REF' and ref($$fields) eq 'ARRAY' ) {
 392     return $self->_fold_sqlbind( $fields );
 393   }
 394   else {
 395     croak($ref . qq{ unexpected in _recurse_fields()})
 396   }
 397 }
 398
 399 sub _order_by {
 400   my ($self, $arg) = @_;
 401
 402   if (ref $arg eq 'HASH' and keys %$arg and not grep { $_ =~ /^-(?:desc|asc)/i } keys %$arg ) {
 403
 404     my $ret = '';
 405
 406     if (my $g = $self->_recurse_fields($arg->{group_by}, { no_rownum_hack => 1 }) ) {
 407       $ret = $self->_sqlcase(' group by ') . $g;
 408     }
 409
 410     if (defined $arg->{having}) {
 411       my ($frag, @bind) = $self->_recurse_where($arg->{having});
 412       push(@{$self->{having_bind}}, @bind);
 413       $ret .= $self->_sqlcase(' having ').$frag;
 414     }
 415
 416     if (defined $arg->{order_by}) {
 417       my ($frag, @bind) = $self->SUPER::_order_by($arg->{order_by});
 418       push(@{$self->{order_bind}}, @bind);
 419       $ret .= $frag;
 420     }
 421
 422     return $ret;
 423   }
 424   else {
 425     my ($sql, @bind) = $self->SUPER::_order_by ($arg);
 426     push(@{$self->{order_bind}}, @bind);
 427     return $sql;
 428   }
 429 }
 430
 431 sub _order_directions {
 432   my ($self, $order) = @_;
 433
 434   # strip bind values - none of the current _order_directions users support them
 435   return $self->SUPER::_order_directions( [ map
 436     { ref $_ ? $_->[0] : $_ }
 437     $self->_order_by_chunks ($order)
 438   ]);
 439 }
 440
 441 sub _table {
 442   my ($self, $from) = @_;
 443   if (ref $from eq 'ARRAY') {
 444     return $self->_recurse_from(@$from);
 445   } elsif (ref $from eq 'HASH') {
 446     return $self->_make_as($from);
 447   } else {
 448     return $from; # would love to quote here but _table ends up getting called
 449                   # twice during an ->select without a limit clause due to
 450                   # the way S::A::Limit->select works. should maybe consider
 451                   # bypassing this and doing S::A::select($self, ...) in
 452                   # our select method above. meantime, quoting shims have
 453                   # been added to select/insert/update/delete here
 454   }
 455 }
 456
 457 sub _recurse_from {
 458   my ($self, $from, @join) = @_;
 459   my @sqlf;
 460   push(@sqlf, $self->_make_as($from));
 461   foreach my $j (@join) {
 462     my ($to, $on) = @$j;
 463
 464
 465     # check whether a join type exists
 466     my $to_jt = ref($to) eq 'ARRAY' ? $to->[0] : $to;
 467     my $join_type;
 468     if (ref($to_jt) eq 'HASH' and defined($to_jt->{-join_type})) {
 469       $join_type = $to_jt->{-join_type};
 470       $join_type =~ s/^\s+ | \s+$//xg;
 471     }
 472
 473     $join_type = $self->{_default_jointype} if not defined $join_type;
 474
 475     my $join_clause = sprintf ('%s JOIN ',
 476       $join_type ?  ' ' . uc($join_type) : ''
 477     );
 478     push @sqlf, $join_clause;
 479
 480     if (ref $to eq 'ARRAY') {
 481       push(@sqlf, '(', $self->_recurse_from(@$to), ')');
 482     } else {
 483       push(@sqlf, $self->_make_as($to));
 484     }
 485     push(@sqlf, ' ON ', $self->_join_condition($on));
 486   }
 487   return join('', @sqlf);
 488 }
 489
 490 sub _fold_sqlbind {
 491   my ($self, $sqlbind) = @_;
 492
 493   my @sqlbind = @$$sqlbind; # copy
 494   my $sql = shift @sqlbind;
 495   push @{$self->{from_bind}}, @sqlbind;
 496
 497   return $sql;
 498 }
 499
 500 sub _make_as {
 501   my ($self, $from) = @_;
 502   return join(' ', map { (ref $_ eq 'SCALAR' ? $$_
 503                         : ref $_ eq 'REF'    ? $self->_fold_sqlbind($_)
 504                         : $self->_quote($_))
 505                        } reverse each %{$self->_skip_options($from)});
 506 }
 507
 508 sub _skip_options {
 509   my ($self, $hash) = @_;
 510   my $clean_hash = {};
 511   $clean_hash->{$_} = $hash->{$_}
 512     for grep {!/^-/} keys %$hash;
 513   return $clean_hash;
 514 }
 515
 516 sub _join_condition {
 517   my ($self, $cond) = @_;
 518   if (ref $cond eq 'HASH') {
 519     my %j;
 520     for (keys %$cond) {
 521       my $v = $cond->{$_};
 522       if (ref $v) {
 523         croak (ref($v) . qq{ reference arguments are not supported in JOINS - try using \"..." instead'})
 524             if ref($v) ne 'SCALAR';
 525         $j{$_} = $v;
 526       }
 527       else {
 528         my $x = '= '.$self->_quote($v); $j{$_} = \$x;
 529       }
 530     };
 531     return scalar($self->_recurse_where(\%j));
 532   } elsif (ref $cond eq 'ARRAY') {
 533     return join(' OR ', map { $self->_join_condition($_) } @$cond);
 534   } else {
 535     die "Can't handle this yet!";
 536   }
 537 }
 538
 539 sub _quote {
 540   my ($self, $label) = @_;
 541   return '' unless defined $label;
 542   return $$label if ref($label) eq 'SCALAR';
 543   return "*" if $label eq '*';
 544   return $label unless $self->{quote_char};
 545   if(ref $self->{quote_char} eq "ARRAY"){
 546     return $self->{quote_char}->[0] . $label . $self->{quote_char}->[1]
 547       if !defined $self->{name_sep};
 548     my $sep = $self->{name_sep};
 549     return join($self->{name_sep},
 550         map { $self->{quote_char}->[0] . $_ . $self->{quote_char}->[1]  }
 551        split(/\Q$sep\E/,$label));
 552   }
 553   return $self->SUPER::_quote($label);
 554 }
 555
 556 sub limit_dialect {
 557     my $self = shift;
 558     $self->{limit_dialect} = shift if @_;
 559     return $self->{limit_dialect};
 560 }
 561
 562 # Set to an array-ref to specify separate left and right quotes for table names.
 563 # A single scalar is equivalen to [ $char, $char ]
 564 sub quote_char {
 565     my $self = shift;
 566     $self->{quote_char} = shift if @_;
 567     return $self->{quote_char};
 568 }
 569
 570 # Character separating quoted table names.
 571 sub name_sep {
 572     my $self = shift;
 573     $self->{name_sep} = shift if @_;
 574     return $self->{name_sep};
 575 }
 576
 577 1;