Preliminary version
[dbsrgits/DBIx-Class.git] / lib / DBIx / Class / SQLAHacks.pm
index 172136b..b5c9f8a 100644 (file)
@@ -46,32 +46,89 @@ sub new {
   $self;
 }
 
+# generate inner/outer select lists for various limit dialects
+# which result in one or more subqueries (e.g. RNO, Top, RowNum)
+# Any non-root-table columns need to have their table qualifier
+# turned into a column name (otherwise names in subqueries clash
+# and/or lose their source table)
+sub _subqueried_selection {
+  my ($self, $rs_attrs) = @_;
+
+  croak 'Limit usable only in the context of DBIC (missing $rs_attrs)' unless $rs_attrs;
+
+  # correlate select and as
+  my @sel;
+  for my $i (0 .. $#{$rs_attrs->{select}}) {
+    my $s = $rs_attrs->{select}[$i];
+    push @sel, {
+      sql => $self->_recurse_fields ($s),
+      unquoted_sql => do { local $self->{quote_char}; $self->_recurse_fields ($s) },
+      as =>
+        ( (ref $s) eq 'HASH' ? $s->{-as} : undef)
+          ||
+        $rs_attrs->{as}[$i]
+          ||
+        croak "Select argument $i ($s) without corresponding 'as'"
+      ,
+    };
+  }
+
+  my ($qsep, $qalias) = map { quotemeta $_ } (
+    $self->name_sep || '.',
+    $rs_attrs->{alias},
+  );
+
+  # re-alias and remove any name separators from aliases,
+  # unless we are dealing with the current source alias
+  # (which will transcend the subqueries and is necessary
+  # for possible further chaining)
+  my (@insel, @outsel);
+  for my $node (@sel) {
+    if (List::Util::first { $_ =~ / (?<! $qalias ) $qsep /x } ($node->{as}, $node->{unquoted_sql}) )  {
+      $node->{as} =~ s/ $qsep /__/xg;
+      push @insel, sprintf '%s AS %s', $node->{sql}, $self->_quote($node->{as});
+      push @outsel, $self->_quote ($node->{as});
+    }
+    else {
+      push @insel, $node->{sql};
+      push @outsel, $self->_quote ($node->{as});
+    }
+  }
+
+  return map { join (', ', @$_ ) } (\@insel, \@outsel);
+}
+
 
 # ANSI standard Limit/Offset implementation. DB2 and MSSQL use this
 sub _RowNumberOver {
-  my ($self, $sql, $order, $rows, $offset ) = @_;
+  my ($self, $sql, $rs_attrs, $rows, $offset ) = @_;
 
-  # get the select to make the final amount of columns equal the original one
-  my ($select) = $sql =~ /^ \s* SELECT \s+ (.+?) \s+ FROM/ix
+  # mangle the input sql as we will be replacing the selector
+  $sql =~ s/^ \s* SELECT \s+ .+? \s+ (?= \b FROM \b )//ix
     or croak "Unrecognizable SELECT: $sql";
 
-  # get the order_by only (or make up an order if none exists)
+  # get selectors
+  my ($insel, $outsel) = $self->_subqueried_selection ($rs_attrs);
+
+  # make up an order if none exists
   my $order_by = $self->_order_by(
-    (delete $order->{order_by}) || $self->_rno_default_order
+    (delete $rs_attrs->{order_by}) || $self->_rno_default_order
   );
 
-  # whatever is left of the order_by
-  my $group_having = $self->_order_by($order);
+  # whatever is left of the order_by (only where is processed at this point)
+  my $group_having = $self->_parse_rs_attrs($rs_attrs);
 
-  my $qalias = $self->_quote ($self->{_dbic_rs_attrs}{alias});
+  my $qalias = $self->_quote ($rs_attrs->{alias});
+
+  my $idx_name = $self->_quote ('rno__row__index');
 
   $sql = sprintf (<<EOS, $offset + 1, $offset + $rows, );
 
-SELECT $select FROM (
-  SELECT $qalias.*, ROW_NUMBER() OVER($order_by ) AS rno__row__index FROM (
-    ${sql}${group_having}
+SELECT $outsel FROM (
+  SELECT $outsel, ROW_NUMBER() OVER($order_by ) AS $idx_name FROM (
+    SELECT $insel ${sql}${group_having}
   ) $qalias
-) $qalias WHERE rno__row__index BETWEEN %d AND %d
+) $qalias WHERE $idx_name BETWEEN %d AND %d
 
 EOS
 
@@ -86,7 +143,7 @@ sub _rno_default_order {
 
 # Informix specific limit, almost like LIMIT/OFFSET
 sub _SkipFirst {
-  my ($self, $sql, $order, $rows, $offset) = @_;
+  my ($self, $sql, $rs_attrs, $rows, $offset) = @_;
 
   $sql =~ s/^ \s* SELECT \s+ //ix
     or croak "Unrecognizable SELECT: $sql";
@@ -98,13 +155,13 @@ sub _SkipFirst {
     ,
     sprintf ('FIRST %d ', $rows),
     $sql,
-    $self->_order_by ($order),
+    $self->_parse_rs_attrs ($rs_attrs),
   );
 }
 
 # Firebird specific limit, reverse of _SkipFirst for Informix
 sub _FirstSkip {
-  my ($self, $sql, $order, $rows, $offset) = @_;
+  my ($self, $sql, $rs_attrs, $rows, $offset) = @_;
 
   $sql =~ s/^ \s* SELECT \s+ //ix
     or croak "Unrecognizable SELECT: $sql";
@@ -116,42 +173,79 @@ sub _FirstSkip {
       : ''
     ,
     $sql,
-    $self->_order_by ($order),
+    $self->_parse_rs_attrs ($rs_attrs),
   );
 }
 
+# WhOracle limits
+sub _RowNum {
+  my ( $self, $sql, $rs_attrs, $rows, $offset ) = @_;
+
+  # mangle the input sql as we will be replacing the selector
+  $sql =~ s/^ \s* SELECT \s+ .+? \s+ (?= \b FROM \b )//ix
+    or croak "Unrecognizable SELECT: $sql";
+
+  my ($insel, $outsel) = $self->_subqueried_selection ($rs_attrs);
+
+  my $qalias = $self->_quote ($rs_attrs->{alias});
+  my $idx_name = $self->_quote ('rownum__index');
+  my $order_group_having = $self->_parse_rs_attrs($rs_attrs);
+
+  $sql = sprintf (<<EOS, $offset + 1, $offset + $rows, );
+
+SELECT $outsel FROM (
+  SELECT $outsel, ROWNUM $idx_name FROM (
+    SELECT $insel ${sql}${order_group_having}
+  ) $qalias
+) $qalias WHERE $idx_name BETWEEN %d AND %d
+
+EOS
+
+  $sql =~ s/\s*\n\s*/ /g;   # easier to read in the debugger
+  return $sql;
+}
+
+=begin
 # Crappy Top based Limit/Offset support. Legacy from MSSQL.
 sub _Top {
-  my ( $self, $sql, $order, $rows, $offset ) = @_;
+  my ( $self, $sql, $rs_attrs, $rows, $offset ) = @_;
 
-  # mangle the input sql so it can be properly aliased in the outer queries
-  $sql =~ s/^ \s* SELECT \s+ (.+?) \s+ (?=FROM)//ix
+  # mangle the input sql as we will be replacing the selector
+  $sql =~ s/^ \s* SELECT \s+ .+? \s+ (?= \b FROM \b )//ix
     or croak "Unrecognizable SELECT: $sql";
-  my $sql_select = $1;
-  my @sql_select = split (/\s*,\s*/, $sql_select);
-
-  # we can't support subqueries (in fact MSSQL can't) - croak
-  if (@sql_select != @{$self->{_dbic_rs_attrs}{select}}) {
-    croak (sprintf (
-      'SQL SELECT did not parse cleanly - retrieved %d comma separated elements, while '
-    . 'the resultset select attribure contains %d elements: %s',
-      scalar @sql_select,
-      scalar @{$self->{_dbic_rs_attrs}{select}},
-      $sql_select,
-    ));
-  }
 
+  # get selectors
+  my ($insel, $outsel) = $self->_subqueried_selection ($rs_attrs);
+
+  # deal with order
+  my $rs_alias = $rs_attrs->{alias};
+  my $req_order = delete $rs_attrs->{order_by};
   my $name_sep = $self->name_sep || '.';
+
+  # examine normalized version, collapses nesting
+  my $limit_order = scalar $self->_order_by_chunks ($req_order)
+    ? $req_order
+    : [ map
+      { join ('', $rs_alias, $name_sep, $_ ) }
+      ( $rs_attrs->{_rsroot_source_handle}->resolve->primary_columns )
+    ]
+  ;
+
+  my ( $order_by_inner, $order_by_outer ) = $self->_order_directions($limit_order);
+  my $order_by_requested = $self->_order_by ($req_order);
+
+
+
+
   my $esc_name_sep = "\Q$name_sep\E";
   my $col_re = qr/ ^ (?: (.+) $esc_name_sep )? ([^$esc_name_sep]+) $ /x;
 
-  my $rs_alias = $self->{_dbic_rs_attrs}{alias};
   my $quoted_rs_alias = $self->_quote ($rs_alias);
 
   # construct the new select lists, rename(alias) some columns if necessary
   my (@outer_select, @inner_select, %seen_names, %col_aliases, %outer_col_aliases);
 
-  for (@{$self->{_dbic_rs_attrs}{select}}) {
+  for (@{$rs_attrs->{select}}) {
     next if ref $_;
     my ($table, $orig_colname) = ( $_ =~ $col_re );
     next unless $table;
@@ -160,7 +254,7 @@ sub _Top {
 
   for my $i (0 .. $#sql_select) {
 
-    my $colsel_arg = $self->{_dbic_rs_attrs}{select}[$i];
+    my $colsel_arg = $rs_attrs->{select}[$i];
     my $colsel_sql = $sql_select[$i];
 
     # this may or may not work (in case of a scalarref or something)
@@ -216,35 +310,14 @@ sub _Top {
 
   %outer_col_aliases = (%outer_col_aliases, %col_aliases);
 
-  # deal with order
-  croak '$order supplied to SQLAHacks limit emulators must be a hash'
-    if (ref $order ne 'HASH');
-
-  $order = { %$order }; #copy
 
-  my $req_order = $order->{order_by};
 
-  # examine normalized version, collapses nesting
-  my $limit_order;
-  if (scalar $self->_order_by_chunks ($req_order)) {
-    $limit_order = $req_order;
-  }
-  else {
-    $limit_order = [ map
-      { join ('', $rs_alias, $name_sep, $_ ) }
-      ( $self->{_dbic_rs_attrs}{_source_handle}->resolve->primary_columns )
-    ];
-  }
-
-  my ( $order_by_inner, $order_by_outer ) = $self->_order_directions($limit_order);
-  my $order_by_requested = $self->_order_by ($req_order);
 
   # generate the rest
-  delete $order->{order_by};
-  my $grpby_having = $self->_order_by ($order);
+  my $grpby_having = $self->_parse_rs_attrs ($rs_attrs);
 
   # short circuit for counts - the ordering complexity is needless
-  if ($self->{_dbic_rs_attrs}{-for_count_only}) {
+  if ($rs_attrs->{-for_count_only}) {
     return "SELECT TOP $rows $inner_select $sql $grpby_having $order_by_outer";
   }
 
@@ -289,29 +362,7 @@ SQL
   $sql =~ s/\s*\n\s*/ /g; # parsing out multiline statements is harder than a single line
   return $sql;
 }
-
-# action at a distance to shorten Top code above
-sub __record_alias {
-  my ($self, $register, $alias, $fqcol, $col) = @_;
-
-  # record qualified name
-  $register->{$fqcol} = $alias;
-  $register->{$self->_quote($fqcol)} = $alias;
-
-  return unless $col;
-
-  # record unqualified name, undef (no adjustment) if a duplicate is found
-  if (exists $register->{$col}) {
-    $register->{$col} = undef;
-  }
-  else {
-    $register->{$col} = $alias;
-  }
-
-  $register->{$self->_quote($col)} = $register->{$col};
-}
-
-
+=cut
 
 # While we're at it, this should make LIMIT queries more efficient,
 #  without digging into things too deeply
@@ -320,14 +371,10 @@ sub _find_syntax {
   return $self->{_cached_syntax} ||= $self->SUPER::_find_syntax($syntax);
 }
 
-my $for_syntax = {
-  update => 'FOR UPDATE',
-  shared => 'FOR SHARE',
-};
 # Quotes table names, handles "limit" dialects (e.g. where rownum between x and
-# y), supports SELECT ... FOR UPDATE and SELECT ... FOR SHARE.
+# y)
 sub select {
-  my ($self, $table, $fields, $where, $order, @rest) = @_;
+  my ($self, $table, $fields, $where, $rs_attrs, @rest) = @_;
 
   $self->{"${_}_bind"} = [] for (qw/having from order/);
 
@@ -340,13 +387,10 @@ sub select {
   @rest = (-1) unless defined $rest[0];
   croak "LIMIT 0 Does Not Compute" if $rest[0] == 0;
     # and anyway, SQL::Abstract::Limit will cause a barf if we don't first
+
   my ($sql, @where_bind) = $self->SUPER::select(
-    $table, $self->_recurse_fields($fields), $where, $order, @rest
+    $table, $self->_recurse_fields($fields), $where, $rs_attrs, @rest
   );
-  if (my $for = delete $self->{_dbic_rs_attrs}{for}) {
-    $sql .= " $for_syntax->{$for}" if $for_syntax->{$for};
-  }
-
   return wantarray ? ($sql, @{$self->{from_bind}}, @where_bind, @{$self->{having_bind}}, @{$self->{order_bind}} ) : $sql;
 }
 
@@ -390,34 +434,35 @@ sub delete {
 
 sub _emulate_limit {
   my $self = shift;
+  # my ( $syntax, $sql, $order, $rows, $offset ) = @_;
+
   if ($_[3] == -1) {
-    return $_[1].$self->_order_by($_[2]);
+    return $_[1] . $self->_parse_rs_attrs($_[2]);
   } else {
     return $self->SUPER::_emulate_limit(@_);
   }
 }
 
 sub _recurse_fields {
-  my ($self, $fields, $params) = @_;
+  my ($self, $fields) = @_;
   my $ref = ref $fields;
   return $self->_quote($fields) unless $ref;
   return $$fields if $ref eq 'SCALAR';
 
   if ($ref eq 'ARRAY') {
-    return join(', ', map {
-      $self->_recurse_fields($_)
-        .(exists $self->{rownum_hack_count} && !($params && $params->{no_rownum_hack})
-          ? ' AS col'.$self->{rownum_hack_count}++
-          : '')
-      } @$fields);
+    return join(', ', map { $self->_recurse_fields($_) } @$fields);
   }
   elsif ($ref eq 'HASH') {
-    my %hash = %$fields;
+    my %hash = %$fields;  # shallow copy
 
     my $as = delete $hash{-as};   # if supplied
 
-    my ($func, $args) = each %hash;
-    delete $hash{$func};
+    my ($func, $args, @toomany) = %hash;
+
+    # there should be only one pair
+    if (@toomany) {
+      croak "Malformed select argument - too many keys in hash: " . join (',', keys %$fields );
+    }
 
     if (lc ($func) eq 'distinct' && ref $args eq 'ARRAY' && @$args > 1) {
       croak (
@@ -435,11 +480,6 @@ sub _recurse_fields {
         : ''
     );
 
-    # there should be nothing left
-    if (keys %hash) {
-      croak "Malformed select argument - too many keys in hash: " . join (',', keys %$fields );
-    }
-
     return $select;
   }
   # Is the second check absolutely necessary?
@@ -451,34 +491,55 @@ sub _recurse_fields {
   }
 }
 
-sub _order_by {
+my $for_syntax = {
+  update => 'FOR UPDATE',
+  shared => 'FOR SHARE',
+};
+
+# this used to be a part of _order_by but is broken out for clarity.
+# What we have been doing forever is hijacking the $order arg of
+# SQLA::select to pass in arbitrary pieces of data (first the group_by,
+# then pretty much the entire resultset attr-hash, as more and more
+# things in the SQLA space need to have mopre info about the $rs they
+# create SQL for. The alternative would be to keep expanding the
+# signature of _select with more and more positional parameters, which
+# is just gross. All hail SQLA2!
+sub _parse_rs_attrs {
   my ($self, $arg) = @_;
 
-  if (ref $arg eq 'HASH' and keys %$arg and not grep { $_ =~ /^-(?:desc|asc)/i } keys %$arg ) {
+  my $sql = '';
 
-    my $ret = '';
+  if (my $g = $self->_recurse_fields($arg->{group_by}, { no_rownum_hack => 1 }) ) {
+    $sql .= $self->_sqlcase(' group by ') . $g;
+  }
 
-    if (my $g = $self->_recurse_fields($arg->{group_by}, { no_rownum_hack => 1 }) ) {
-      $ret = $self->_sqlcase(' group by ') . $g;
-    }
+  if (defined $arg->{having}) {
+    my ($frag, @bind) = $self->_recurse_where($arg->{having});
+    push(@{$self->{having_bind}}, @bind);
+    $sql .= $self->_sqlcase(' having ') . $frag;
+  }
 
-    if (defined $arg->{having}) {
-      my ($frag, @bind) = $self->_recurse_where($arg->{having});
-      push(@{$self->{having_bind}}, @bind);
-      $ret .= $self->_sqlcase(' having ').$frag;
-    }
+  if (defined $arg->{order_by}) {
+    $sql .= $self->_order_by ($arg->{order_by});
+  }
 
-    if (defined $arg->{order_by}) {
-      my ($frag, @bind) = $self->SUPER::_order_by($arg->{order_by});
-      push(@{$self->{order_bind}}, @bind);
-      $ret .= $frag;
-    }
+  if (my $for = $arg->{for}) {
+    $sql .= " $for_syntax->{$for}" if $for_syntax->{$for};
+  }
+
+  return $sql;
+}
+
+sub _order_by {
+  my ($self, $arg) = @_;
 
-    return $ret;
+  # check that we are not called in legacy mode (order_by as 4th argument)
+  if (ref $arg eq 'HASH' and not grep { $_ =~ /^-(?:desc|asc)/i } keys %$arg ) {
+    return $self->_parse_rs_attrs ($arg);
   }
   else {
     my ($sql, @bind) = $self->SUPER::_order_by ($arg);
-    push(@{$self->{order_bind}}, @bind);
+    push @{$self->{order_bind}}, @bind;
     return $sql;
   }
 }
@@ -598,7 +659,10 @@ sub _join_condition {
 
 sub limit_dialect {
     my $self = shift;
-    $self->{limit_dialect} = shift if @_;
+    if (@_) {
+      $self->{limit_dialect} = shift;
+      undef $self->{_cached_syntax};
+    }
     return $self->{limit_dialect};
 }