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