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