lib/DBIx/Class/SQLAHacks/LimitDialects.pm

   1 package DBIx::Class::SQLAHacks::LimitDialects;
   2
   3 use warnings;
   4 use strict;
   5
   6 use Carp::Clan qw/^DBIx::Class|^SQL::Abstract|^Try::Tiny/;
   7 use List::Util 'first';
   8 use namespace::clean;
   9
  10 # PostgreSQL and SQLite
  11 sub _LimitOffset {
  12     my ( $self, $sql, $order, $rows, $offset ) = @_;
  13     $sql .= $self->_order_by( $order ) . " LIMIT $rows";
  14     $sql .= " OFFSET $offset" if +$offset;
  15     return $sql;
  16 }
  17
  18 # MySQL and any SQL::Statement based DBD
  19 sub _LimitXY {
  20     my ( $self, $sql, $order, $rows, $offset ) = @_;
  21     $sql .= $self->_order_by( $order ) . " LIMIT ";
  22     $sql .= "$offset, " if +$offset;
  23     $sql .= $rows;
  24     return $sql;
  25 }
  26 # ANSI standard Limit/Offset implementation. DB2 and MSSQL >= 2005 use this
  27 sub _RowNumberOver {
  28   my ($self, $sql, $rs_attrs, $rows, $offset ) = @_;
  29
  30   # mangle the input sql as we will be replacing the selector
  31   $sql =~ s/^ \s* SELECT \s+ .+? \s+ (?= \b FROM \b )//ix
  32     or croak "Unrecognizable SELECT: $sql";
  33
  34   # get selectors, and scan the order_by (if any)
  35   my ($in_sel, $out_sel, $alias_map, $extra_order_sel)
  36     = $self->_subqueried_limit_attrs ( $rs_attrs );
  37
  38   # make up an order if none exists
  39   my $requested_order = (delete $rs_attrs->{order_by}) || $self->_rno_default_order;
  40   my $rno_ord = $self->_order_by ($requested_order);
  41
  42   # this is the order supplement magic
  43   my $mid_sel = $out_sel;
  44   if ($extra_order_sel) {
  45     for my $extra_col (sort
  46       { $extra_order_sel->{$a} cmp $extra_order_sel->{$b} }
  47       keys %$extra_order_sel
  48     ) {
  49       $in_sel .= sprintf (', %s AS %s',
  50         $extra_col,
  51         $extra_order_sel->{$extra_col},
  52       );
  53
  54       $mid_sel .= ', ' . $extra_order_sel->{$extra_col};
  55     }
  56   }
  57
  58   # and this is order re-alias magic
  59   for ($extra_order_sel, $alias_map) {
  60     for my $col (keys %$_) {
  61       my $re_col = quotemeta ($col);
  62       $rno_ord =~ s/$re_col/$_->{$col}/;
  63     }
  64   }
  65
  66   # whatever is left of the order_by (only where is processed at this point)
  67   my $group_having = $self->_parse_rs_attrs($rs_attrs);
  68
  69   my $qalias = $self->_quote ($rs_attrs->{alias});
  70   my $idx_name = $self->_quote ('rno__row__index');
  71
  72   $sql = sprintf (<<EOS, $offset + 1, $offset + $rows, );
  73
  74 SELECT $out_sel FROM (
  75   SELECT $mid_sel, ROW_NUMBER() OVER( $rno_ord ) AS $idx_name FROM (
  76     SELECT $in_sel ${sql}${group_having}
  77   ) $qalias
  78 ) $qalias WHERE $idx_name BETWEEN %u AND %u
  79
  80 EOS
  81
  82   $sql =~ s/\s*\n\s*/ /g;   # easier to read in the debugger
  83   return $sql;
  84 }
  85
  86 # some databases are happy with OVER (), some need OVER (ORDER BY (SELECT (1)) )
  87 sub _rno_default_order {
  88   return undef;
  89 }
  90
  91 # Informix specific limit, almost like LIMIT/OFFSET
  92 # According to SQLA::Limit informix also supports
  93 # SKIP X LIMIT Y (in addition to SKIP X FIRST Y)
  94 sub _SkipFirst {
  95   my ($self, $sql, $rs_attrs, $rows, $offset) = @_;
  96
  97   $sql =~ s/^ \s* SELECT \s+ //ix
  98     or croak "Unrecognizable SELECT: $sql";
  99
 100   return sprintf ('SELECT %s%s%s%s',
 101     $offset
 102       ? sprintf ('SKIP %u ', $offset)
 103       : ''
 104     ,
 105     sprintf ('FIRST %u ', $rows),
 106     $sql,
 107     $self->_parse_rs_attrs ($rs_attrs),
 108   );
 109 }
 110
 111 # Firebird specific limit, reverse of _SkipFirst for Informix
 112 # According to SQLA::Limit firebird/interbase also supports
 113 # ROWS X TO Y (in addition to FIRST X SKIP Y)
 114 sub _FirstSkip {
 115   my ($self, $sql, $rs_attrs, $rows, $offset) = @_;
 116
 117   $sql =~ s/^ \s* SELECT \s+ //ix
 118     or croak "Unrecognizable SELECT: $sql";
 119
 120   return sprintf ('SELECT %s%s%s%s',
 121     sprintf ('FIRST %u ', $rows),
 122     $offset
 123       ? sprintf ('SKIP %u ', $offset)
 124       : ''
 125     ,
 126     $sql,
 127     $self->_parse_rs_attrs ($rs_attrs),
 128   );
 129 }
 130
 131 # WhOracle limits
 132 sub _RowNum {
 133   my ( $self, $sql, $rs_attrs, $rows, $offset ) = @_;
 134
 135   # mangle the input sql as we will be replacing the selector
 136   $sql =~ s/^ \s* SELECT \s+ .+? \s+ (?= \b FROM \b )//ix
 137     or croak "Unrecognizable SELECT: $sql";
 138
 139   my ($insel, $outsel) = $self->_subqueried_limit_attrs ($rs_attrs);
 140
 141   my $qalias = $self->_quote ($rs_attrs->{alias});
 142   my $idx_name = $self->_quote ('rownum__index');
 143   my $order_group_having = $self->_parse_rs_attrs($rs_attrs);
 144
 145   $sql = sprintf (<<EOS, $offset + 1, $offset + $rows, );
 146
 147 SELECT $outsel FROM (
 148   SELECT $outsel, ROWNUM $idx_name FROM (
 149     SELECT $insel ${sql}${order_group_having}
 150   ) $qalias
 151 ) $qalias WHERE $idx_name BETWEEN %u AND %u
 152
 153 EOS
 154
 155   $sql =~ s/\s*\n\s*/ /g;   # easier to read in the debugger
 156   return $sql;
 157 }
 158
 159 # Crappy Top based Limit/Offset support. Legacy for MSSQL < 2005
 160 sub _Top {
 161   my ( $self, $sql, $rs_attrs, $rows, $offset ) = @_;
 162
 163   # mangle the input sql as we will be replacing the selector
 164   $sql =~ s/^ \s* SELECT \s+ .+? \s+ (?= \b FROM \b )//ix
 165     or croak "Unrecognizable SELECT: $sql";
 166
 167   # get selectors
 168   my ($in_sel, $out_sel, $alias_map, $extra_order_sel)
 169     = $self->_subqueried_limit_attrs ($rs_attrs);
 170
 171   my $requested_order = delete $rs_attrs->{order_by};
 172
 173   my $order_by_requested = $self->_order_by ($requested_order);
 174
 175   # make up an order unless supplied
 176   my $inner_order = ($order_by_requested
 177     ? $requested_order
 178     : [ map
 179       { join ('', $rs_attrs->{alias}, $self->{name_sep}||'.', $_ ) }
 180       ( $rs_attrs->{_rsroot_source_handle}->resolve->_pri_cols )
 181     ]
 182   );
 183
 184   my ($order_by_inner, $order_by_reversed);
 185
 186   # localise as we already have all the bind values we need
 187   {
 188     local $self->{order_bind};
 189     $order_by_inner = $self->_order_by ($inner_order);
 190
 191     my @out_chunks;
 192     for my $ch ($self->_order_by_chunks ($inner_order)) {
 193       $ch = $ch->[0] if ref $ch eq 'ARRAY';
 194
 195       $ch =~ s/\s+ ( ASC|DESC ) \s* $//ix;
 196       my $dir = uc ($1||'ASC');
 197
 198       push @out_chunks, \join (' ', $ch, $dir eq 'ASC' ? 'DESC' : 'ASC' );
 199     }
 200
 201     $order_by_reversed = $self->_order_by (\@out_chunks);
 202   }
 203
 204   # this is the order supplement magic
 205   my $mid_sel = $out_sel;
 206   if ($extra_order_sel) {
 207     for my $extra_col (sort
 208       { $extra_order_sel->{$a} cmp $extra_order_sel->{$b} }
 209       keys %$extra_order_sel
 210     ) {
 211       $in_sel .= sprintf (', %s AS %s',
 212         $extra_col,
 213         $extra_order_sel->{$extra_col},
 214       );
 215
 216       $mid_sel .= ', ' . $extra_order_sel->{$extra_col};
 217     }
 218
 219     # since whatever order bindvals there are, they will be realiased
 220     # and need to show up in front of the entire initial inner subquery
 221     # Unshift *from_bind* to make this happen (horrible, horrible, but
 222     # we don't have another mechanism yet)
 223     unshift @{$self->{from_bind}}, @{$self->{order_bind}};
 224   }
 225
 226   # and this is order re-alias magic
 227   for my $map ($extra_order_sel, $alias_map) {
 228     for my $col (keys %$map) {
 229       my $re_col = quotemeta ($col);
 230       $_ =~ s/$re_col/$map->{$col}/
 231         for ($order_by_reversed, $order_by_requested);
 232     }
 233   }
 234
 235   # generate the rest of the sql
 236   my $grpby_having = $self->_parse_rs_attrs ($rs_attrs);
 237
 238   my $quoted_rs_alias = $self->_quote ($rs_attrs->{alias});
 239
 240   $sql = sprintf ('SELECT TOP %u %s %s %s %s',
 241     $rows + ($offset||0),
 242     $in_sel,
 243     $sql,
 244     $grpby_having,
 245     $order_by_inner,
 246   );
 247
 248   $sql = sprintf ('SELECT TOP %u %s FROM ( %s ) %s %s',
 249     $rows,
 250     $mid_sel,
 251     $sql,
 252     $quoted_rs_alias,
 253     $order_by_reversed,
 254   ) if $offset;
 255
 256   $sql = sprintf ('SELECT TOP %u %s FROM ( %s ) %s %s',
 257     $rows,
 258     $out_sel,
 259     $sql,
 260     $quoted_rs_alias,
 261     $order_by_requested,
 262   ) if ( ($offset && $order_by_requested) || ($mid_sel ne $out_sel) );
 263
 264   $sql =~ s/\s*\n\s*/ /g;   # easier to read in the debugger
 265   return $sql;
 266 }
 267
 268 # This for Sybase ASE, to use SET ROWCOUNT when there is no offset, and
 269 # GenericSubQ otherwise.
 270 sub _RowCountOrGenericSubQ {
 271   my $self = shift;
 272   my ($sql, $rs_attrs, $rows, $offset) = @_;
 273
 274   return $self->_GenericSubQ(@_) if $offset;
 275
 276   return sprintf <<"EOF", $rows, $sql;
 277 SET ROWCOUNT %d
 278 %s
 279 SET ROWCOUNT 0
 280 EOF
 281 }
 282
 283 # This is the most evil limit "dialect" (more of a hack) for *really*
 284 # stupid databases. It works by ordering the set by some unique column,
 285 # and calculating amount of rows that have a less-er value (thus
 286 # emulating a RowNum-like index). Of course this implies the set can
 287 # only be ordered by a single unique columns.
 288 sub _GenericSubQ {
 289   my ($self, $sql, $rs_attrs, $rows, $offset) = @_;
 290
 291   my $root_rsrc = $rs_attrs->{_rsroot_source_handle}->resolve;
 292   my $root_tbl_name = $root_rsrc->name;
 293
 294   # mangle the input sql as we will be replacing the selector
 295   $sql =~ s/^ \s* SELECT \s+ .+? \s+ (?= \b FROM \b )//ix
 296     or croak "Unrecognizable SELECT: $sql";
 297
 298   my ($order_by, @rest) = do {
 299     local $self->{quote_char};
 300     $self->_order_by_chunks ($rs_attrs->{order_by})
 301   };
 302
 303   unless (
 304     $order_by
 305       &&
 306     ! @rest
 307       &&
 308     ( ! ref $order_by
 309         ||
 310       ( ref $order_by eq 'ARRAY' and @$order_by == 1 )
 311     )
 312   ) {
 313     croak (
 314       'Generic Subquery Limit does not work on resultsets without an order, or resultsets '
 315     . 'with complex order criteria (multicolumn and/or functions). Provide a single, '
 316     . 'unique-column order criteria.'
 317     );
 318   }
 319
 320   ($order_by) = @$order_by if ref $order_by;
 321
 322   $order_by =~ s/\s+ ( ASC|DESC ) \s* $//ix;
 323   my $direction = lc ($1 || 'asc');
 324
 325   my ($unq_sort_col) = $order_by =~ /(?:^|\.)([^\.]+)$/;
 326
 327   my $inf = $root_rsrc->storage->_resolve_column_info (
 328     $rs_attrs->{from}, [$order_by, $unq_sort_col]
 329   );
 330
 331   my $ord_colinfo = $inf->{$order_by} || croak "Unable to determine source of order-criteria '$order_by'";
 332
 333   if ($ord_colinfo->{-result_source}->name ne $root_tbl_name) {
 334     croak "Generic Subquery Limit order criteria can be only based on the root-source '"
 335         . $root_rsrc->source_name . "' (aliased as '$rs_attrs->{alias}')";
 336   }
 337
 338   # make sure order column is qualified
 339   $order_by = "$rs_attrs->{alias}.$order_by"
 340     unless $order_by =~ /^$rs_attrs->{alias}\./;
 341
 342   my $is_u;
 343   my $ucs = { $root_rsrc->unique_constraints };
 344   for (values %$ucs ) {
 345     if (@$_ == 1 && "$rs_attrs->{alias}.$_->[0]" eq $order_by) {
 346       $is_u++;
 347       last;
 348     }
 349   }
 350   croak "Generic Subquery Limit order criteria column '$order_by' must be unique (no unique constraint found)"
 351     unless $is_u;
 352
 353   my ($in_sel, $out_sel, $alias_map, $extra_order_sel)
 354     = $self->_subqueried_limit_attrs ($rs_attrs);
 355
 356   my $cmp_op = $direction eq 'desc' ? '>' : '<';
 357   my $count_tbl_alias = 'rownum__emulation';
 358
 359   my $order_sql = $self->_order_by (delete $rs_attrs->{order_by});
 360   my $group_having_sql = $self->_parse_rs_attrs($rs_attrs);
 361
 362   # add the order supplement (if any) as this is what will be used for the outer WHERE
 363   $in_sel .= ", $_" for keys %{$extra_order_sel||{}};
 364
 365   $sql = sprintf (<<EOS,
 366 SELECT $out_sel
 367   FROM (
 368     SELECT $in_sel ${sql}${group_having_sql}
 369   ) %s
 370 WHERE ( SELECT COUNT(*) FROM %s %s WHERE %s $cmp_op %s ) %s
 371 $order_sql
 372 EOS
 373     ( map { $self->_quote ($_) } (
 374       $rs_attrs->{alias},
 375       $root_tbl_name,
 376       $count_tbl_alias,
 377       "$count_tbl_alias.$unq_sort_col",
 378       $order_by,
 379     )),
 380     $offset
 381       ? sprintf ('BETWEEN %u AND %u', $offset, $offset + $rows - 1)
 382       : sprintf ('< %u', $rows )
 383     ,
 384   );
 385
 386   $sql =~ s/\s*\n\s*/ /g;   # easier to read in the debugger
 387   return $sql;
 388 }
 389
 390
 391 # !!! THIS IS ALSO HORRIFIC !!! /me ashamed
 392 #
 393 # Generates inner/outer select lists for various limit dialects
 394 # which result in one or more subqueries (e.g. RNO, Top, RowNum)
 395 # Any non-root-table columns need to have their table qualifier
 396 # turned into a column alias (otherwise names in subqueries clash
 397 # and/or lose their source table)
 398 #
 399 # Returns inner/outer strings of SQL QUOTED selectors with aliases
 400 # (to be used in whatever select statement), and an alias index hashref
 401 # of QUOTED SEL => QUOTED ALIAS pairs (to maybe be used for string-subst
 402 # higher up).
 403 # If an order_by is supplied, the inner select needs to bring out columns
 404 # used in implicit (non-selected) orders, and the order condition itself
 405 # needs to be realiased to the proper names in the outer query. Thus we
 406 # also return a hashref (order doesn't matter) of QUOTED EXTRA-SEL =>
 407 # QUOTED ALIAS pairs, which is a list of extra selectors that do *not*
 408 # exist in the original select list
 409
 410 sub _subqueried_limit_attrs {
 411   my ($self, $rs_attrs) = @_;
 412
 413   croak 'Limit dialect implementation usable only in the context of DBIC (missing $rs_attrs)'
 414     unless ref ($rs_attrs) eq 'HASH';
 415
 416   my ($re_sep, $re_alias) = map { quotemeta $_ } (
 417     $self->name_sep || '.',
 418     $rs_attrs->{alias},
 419   );
 420
 421   # correlate select and as, build selection index
 422   my (@sel, $in_sel_index);
 423   for my $i (0 .. $#{$rs_attrs->{select}}) {
 424
 425     my $s = $rs_attrs->{select}[$i];
 426     my $sql_sel = $self->_recurse_fields ($s);
 427     my $sql_alias = (ref $s) eq 'HASH' ? $s->{-as} : undef;
 428
 429
 430     push @sel, {
 431       sql => $sql_sel,
 432       unquoted_sql => do { local $self->{quote_char}; $self->_recurse_fields ($s) },
 433       as =>
 434         $sql_alias
 435           ||
 436         $rs_attrs->{as}[$i]
 437           ||
 438         croak "Select argument $i ($s) without corresponding 'as'"
 439       ,
 440     };
 441
 442     $in_sel_index->{$sql_sel}++;
 443     $in_sel_index->{$self->_quote ($sql_alias)}++ if $sql_alias;
 444
 445     # record unqualified versions too, so we do not have
 446     # to reselect the same column twice (in qualified and
 447     # unqualified form)
 448     if (! ref $s && $sql_sel =~ / $re_sep (.+) $/x) {
 449       $in_sel_index->{$1}++;
 450     }
 451   }
 452
 453
 454   # re-alias and remove any name separators from aliases,
 455   # unless we are dealing with the current source alias
 456   # (which will transcend the subqueries as it is necessary
 457   # for possible further chaining)
 458   my (@in_sel, @out_sel, %renamed);
 459   for my $node (@sel) {
 460     if (first { $_ =~ / (?<! ^ $re_alias ) $re_sep /x } ($node->{as}, $node->{unquoted_sql}) )  {
 461       $node->{as} = $self->_unqualify_colname($node->{as});
 462       my $quoted_as = $self->_quote($node->{as});
 463       push @in_sel, sprintf '%s AS %s', $node->{sql}, $quoted_as;
 464       push @out_sel, $quoted_as;
 465       $renamed{$node->{sql}} = $quoted_as;
 466     }
 467     else {
 468       push @in_sel, $node->{sql};
 469       push @out_sel, $self->_quote ($node->{as});
 470     }
 471   }
 472
 473   # see if the order gives us anything
 474   my %extra_order_sel;
 475   for my $chunk ($self->_order_by_chunks ($rs_attrs->{order_by})) {
 476     # order with bind
 477     $chunk = $chunk->[0] if (ref $chunk) eq 'ARRAY';
 478     $chunk =~ s/\s+ (?: ASC|DESC ) \s* $//ix;
 479
 480     next if $in_sel_index->{$chunk};
 481
 482     $extra_order_sel{$chunk} ||= $self->_quote (
 483       'ORDER__BY__' . scalar keys %extra_order_sel
 484     );
 485   }
 486
 487   return (
 488     (map { join (', ', @$_ ) } (
 489       \@in_sel,
 490       \@out_sel)
 491     ),
 492     \%renamed,
 493     keys %extra_order_sel ? \%extra_order_sel : (),
 494   );
 495 }
 496
 497 sub _unqualify_colname {
 498   my ($self, $fqcn) = @_;
 499   my $re_sep = quotemeta($self->name_sep || '.');
 500   $fqcn =~ s/ $re_sep /__/xg;
 501   return $fqcn;
 502 }
 503
 504 1;