DBIx::Class::ResultSource - Result source object

  # Create a table based result source, in a result class.

  package MyApp::Schema::Result::Artist;
  use base qw/DBIx::Class::Core/;

  __PACKAGE__->table('artist');
  __PACKAGE__->add_columns(qw/ artistid name /);
  __PACKAGE__->set_primary_key('artistid');
  __PACKAGE__->has_many(cds => 'MyApp::Schema::Result::CD');

  1;

  # Create a query (view) based result source, in a result class
  package MyApp::Schema::Result::Year2000CDs;
  use base qw/DBIx::Class::Core/;

  __PACKAGE__->load_components('InflateColumn::DateTime');
  __PACKAGE__->table_class('DBIx::Class::ResultSource::View');

  __PACKAGE__->table('year2000cds');
  __PACKAGE__->result_source_instance->is_virtual(1);
  __PACKAGE__->result_source_instance->view_definition(
      "SELECT cdid, artist, title FROM cd WHERE year ='2000'"
      );

  $source->add_columns(qw/col1 col2 col3/);

  $source->add_columns('col1' => \%col1_info, 'col2' => \%col2_info, ...);

  $source->add_column('col' => \%info);

  if ($source->has_column($colname)) { ... }

  my $info = $source->column_info($col);

  my @column_names = $source->columns;

  my $columns_info = $source->columns_info;

  $source->remove_columns(qw/col1 col2 col3/);

  $source->remove_column('col');

  # For UNIQUE (column1, column2)
  __PACKAGE__->add_unique_constraint(
    constraint_name => [ qw/column1 column2/ ],
  );

  __PACKAGE__->add_unique_constraint([ qw/column1 column2/ ]);

  __PACKAGE__->add_unique_constraints(
    constraint_name1 => [ qw/column1 column2/ ],
    constraint_name2 => [ qw/column2 column3/ ],
  );

  __PACKAGE__->add_unique_constraints(
    [ qw/column1 column2/ ],
    [ qw/column3 column4/ ]
  );

  $source->table('mytable');
  $source->name_unique_constraint(['col1', 'col2']);
  # returns
  'mytable_col1_col2'

  $source->unique_constraints();

  $source->unique_constraint_names();

  $source->unique_constraint_columns('myconstraint');

  __PACKAGE__->sqlt_deploy_callback('mycallbackmethod');

  $self->resultset_class->new($self, $self->resultset_attributes)

  package My::Schema::ResultSet::Artist;
  use base 'DBIx::Class::ResultSet';
  ...

  # In the result class
  __PACKAGE__->resultset_class('My::Schema::ResultSet::Artist');

  # Or in code
  $source->resultset_class('My::Schema::ResultSet::Artist');

  # In the result class
  __PACKAGE__->resultset_attributes({ order_by => [ 'id' ] });

  # Or in code
  $source->resultset_attributes({ order_by => [ 'id' ] });

  package ArchivedBooks;
  use base qw/DBIx::Class/;
  __PACKAGE__->table('books_archive');
  __PACKAGE__->source_name('Books');

  # from your schema...
  $schema->resultset('Books')->find(1);

  my $from_clause = $source->from();

  my $schema = $source->schema();

  $source->storage->debug(1);

  $source->add_relationship('relname', 'related_source', $cond, $attrs);

  $schema->source('Book')->add_relationship('reviews', 'Review', {
    'foreign.book_id' => 'self.id',
  });

  { 'foreign.author_id' => 'self.id' }

  author me JOIN book foreign ON foreign.author_id = me.id

  my @relnames = $source->relationships();

  __PACKAGE__->source_info({
    "_tablespace" => 'fast_disk_array_3',
    "_engine" => 'InnoDB',
  });

  $class->new();

  $class->new({attribute_name => value});

  __PACKAGE__->column_info_from_storage(1);

package DBIx::Class::ResultSource;

use strict;
use warnings;

use DBIx::Class::ResultSet;
use DBIx::Class::ResultSourceHandle;

use DBIx::Class::Exception;
use DBIx::Class::Carp;
use Try::Tiny;
use List::Util 'first';
use Scalar::Util qw/blessed weaken isweak/;
use namespace::clean;

use base qw/DBIx::Class/;

__PACKAGE__->mk_group_accessors('simple' => qw/_ordered_columns
  _columns _primaries _unique_constraints name resultset_attributes
  from _relationships column_info_from_storage source_info
  source_name sqlt_deploy_callback/);

__PACKAGE__->mk_group_accessors('component_class' => qw/resultset_class
  result_class/);

sub new {
  my ($class, $attrs) = @_;
  $class = ref $class if ref $class;

  my $new = bless { %{$attrs || {}} }, $class;
  $new->{resultset_class} ||= 'DBIx::Class::ResultSet';
  $new->{resultset_attributes} = { %{$new->{resultset_attributes} || {}} };
  $new->{_ordered_columns} = [ @{$new->{_ordered_columns}||[]}];
  $new->{_columns} = { %{$new->{_columns}||{}} };
  $new->{_relationships} = { %{$new->{_relationships}||{}} };
  $new->{name} ||= "!!NAME NOT SET!!";
  $new->{_columns_info_loaded} ||= 0;
  $new->{sqlt_deploy_callback} ||= "default_sqlt_deploy_hook";
  return $new;
}

sub add_columns {
  my ($self, @cols) = @_;
  $self->_ordered_columns(\@cols) unless $self->_ordered_columns;

  my @added;
  my $columns = $self->_columns;
  while (my $col = shift @cols) {
    my $column_info = {};
    if ($col =~ s/^\+//) {
      $column_info = $self->column_info($col);
    }

    # If next entry is { ... } use that for the column info, if not
    # use an empty hashref
    if (ref $cols[0]) {
      my $new_info = shift(@cols);
      %$column_info = (%$column_info, %$new_info);
    }
    push(@added, $col) unless exists $columns->{$col};
    $columns->{$col} = $column_info;
  }
  push @{ $self->_ordered_columns }, @added;
  return $self;
}

sub add_column { shift->add_columns(@_); } # DO NOT CHANGE THIS TO GLOB

sub has_column {
  my ($self, $column) = @_;
  return exists $self->_columns->{$column};
}

sub column_info {
  my ($self, $column) = @_;
  $self->throw_exception("No such column $column")
    unless exists $self->_columns->{$column};

  if ( ! $self->_columns->{$column}{data_type}
       and ! $self->{_columns_info_loaded}
       and $self->column_info_from_storage
       and my $stor = try { $self->storage } )
  {
    $self->{_columns_info_loaded}++;

    # try for the case of storage without table
    try {
      my $info = $stor->columns_info_for( $self->from );
      my $lc_info = { map
        { (lc $_) => $info->{$_} }
        ( keys %$info )
      };

      foreach my $col ( keys %{$self->_columns} ) {
        $self->_columns->{$col} = {
          %{ $self->_columns->{$col} },
          %{ $info->{$col} || $lc_info->{lc $col} || {} }
        };
      }
    };
  }

  return $self->_columns->{$column};
}

sub columns {
  my $self = shift;
  $self->throw_exception(
    "columns() is a read-only accessor, did you mean add_columns()?"
  ) if @_;
  return @{$self->{_ordered_columns}||[]};
}

sub columns_info {
  my ($self, $columns) = @_;

  my $colinfo = $self->_columns;

  if (
    first { ! $_->{data_type} } values %$colinfo
      and
    ! $self->{_columns_info_loaded}
      and
    $self->column_info_from_storage
      and
    my $stor = try { $self->storage }
  ) {
    $self->{_columns_info_loaded}++;

    # try for the case of storage without table
    try {
      my $info = $stor->columns_info_for( $self->from );
      my $lc_info = { map
        { (lc $_) => $info->{$_} }
        ( keys %$info )
      };

      foreach my $col ( keys %$colinfo ) {
        $colinfo->{$col} = {
          %{ $colinfo->{$col} },
          %{ $info->{$col} || $lc_info->{lc $col} || {} }
        };
      }
    };
  }

  my %ret;

  if ($columns) {
    for (@$columns) {
      if (my $inf = $colinfo->{$_}) {
        $ret{$_} = $inf;
      }
      else {
        $self->throw_exception( sprintf (
          "No such column '%s' on source %s",
          $_,
          $self->source_name,
        ));
      }
    }
  }
  else {
    %ret = %$colinfo;
  }

  return \%ret;
}

sub remove_columns {
  my ($self, @to_remove) = @_;

  my $columns = $self->_columns
    or return;

  my %to_remove;
  for (@to_remove) {
    delete $columns->{$_};
    ++$to_remove{$_};
  }

  $self->_ordered_columns([ grep { not $to_remove{$_} } @{$self->_ordered_columns} ]);
}

sub remove_column { shift->remove_columns(@_); } # DO NOT CHANGE THIS TO GLOB

sub set_primary_key {
  my ($self, @cols) = @_;
  # check if primary key columns are valid columns
  foreach my $col (@cols) {
    $self->throw_exception("No such column $col on table " . $self->name)
      unless $self->has_column($col);
  }
  $self->_primaries(\@cols);

  $self->add_unique_constraint(primary => \@cols);
}

sub primary_columns {
  return @{shift->_primaries||[]};
}

# a helper method that will automatically die with a descriptive message if
# no pk is defined on the source in question. For internal use to save
# on if @pks... boilerplate
sub _pri_cols {
  my $self = shift;
  my @pcols = $self->primary_columns
    or $self->throw_exception (sprintf(
      "Operation requires a primary key to be declared on '%s' via set_primary_key",
      # source_name is set only after schema-registration
      $self->source_name || $self->result_class || $self->name || 'Unknown source...?',
    ));
  return @pcols;
}

sub sequence {
  my ($self,$seq) = @_;

  my @pks = $self->primary_columns
    or return;

  $_->{sequence} = $seq
    for values %{ $self->columns_info (\@pks) };
}

sub add_unique_constraint {
  my $self = shift;

  if (@_ > 2) {
    $self->throw_exception(
        'add_unique_constraint() does not accept multiple constraints, use '
      . 'add_unique_constraints() instead'
    );
  }

  my $cols = pop @_;
  if (ref $cols ne 'ARRAY') {
    $self->throw_exception (
      'Expecting an arrayref of constraint columns, got ' . ($cols||'NOTHING')
    );
  }

  my $name = shift @_;

  $name ||= $self->name_unique_constraint($cols);

  foreach my $col (@$cols) {
    $self->throw_exception("No such column $col on table " . $self->name)
      unless $self->has_column($col);
  }

  my %unique_constraints = $self->unique_constraints;
  $unique_constraints{$name} = $cols;
  $self->_unique_constraints(\%unique_constraints);
}

sub add_unique_constraints {
  my $self = shift;
  my @constraints = @_;

  if ( !(@constraints % 2) && first { ref $_ ne 'ARRAY' } @constraints ) {
    # with constraint name
    while (my ($name, $constraint) = splice @constraints, 0, 2) {
      $self->add_unique_constraint($name => $constraint);
    }
  }
  else {
    # no constraint name
    foreach my $constraint (@constraints) {
      $self->add_unique_constraint($constraint);
    }
  }
}

sub name_unique_constraint {
  my ($self, $cols) = @_;

  my $name = $self->name;
  $name = $$name if (ref $name eq 'SCALAR');

  return join '_', $name, @$cols;
}

sub unique_constraints {
  return %{shift->_unique_constraints||{}};
}

sub unique_constraint_names {
  my ($self) = @_;

  my %unique_constraints = $self->unique_constraints;

  return keys %unique_constraints;
}

sub unique_constraint_columns {
  my ($self, $constraint_name) = @_;

  my %unique_constraints = $self->unique_constraints;

  $self->throw_exception(
    "Unknown unique constraint $constraint_name on '" . $self->name . "'"
  ) unless exists $unique_constraints{$constraint_name};

  return @{ $unique_constraints{$constraint_name} };
}

sub default_sqlt_deploy_hook {
  my $self = shift;

  my $class = $self->result_class;

  if ($class and my $hook = $class->can('sqlt_deploy_hook')) {
    $self->$hook(@_);
  }
}

sub _invoke_sqlt_deploy_hook {
  my $self = shift;
  if ( my $hook = $self->sqlt_deploy_callback) {
    $self->$hook(@_);
  }
}

sub resultset {
  my $self = shift;
  $self->throw_exception(
    'resultset does not take any arguments. If you want another resultset, '.
    'call it on the schema instead.'
  ) if scalar @_;

  $self->resultset_class->new(
    $self,
    {
      try { %{$self->schema->default_resultset_attributes} },
      %{$self->{resultset_attributes}},
    },
  );
}

sub schema {
  if (@_ > 1) {
    $_[0]->{schema} = $_[1];
  }
  else {
    $_[0]->{schema} || do {
      my $name = $_[0]->{source_name} || '_unnamed_';
      my $err = 'Unable to perform storage-dependent operations with a detached result source '
              . "(source '$name' is not associated with a schema).";

      $err .= ' You need to use $schema->thaw() or manually set'
            . ' $DBIx::Class::ResultSourceHandle::thaw_schema while thawing.'
        if $_[0]->{_detached_thaw};

      DBIx::Class::Exception->throw($err);
    };
  }
}

sub storage { shift->schema->storage; }

sub add_relationship {
  my ($self, $rel, $f_source_name, $cond, $attrs) = @_;
  $self->throw_exception("Can't create relationship without join condition")
    unless $cond;
  $attrs ||= {};

  # Check foreign and self are right in cond
  if ( (ref $cond ||'') eq 'HASH') {
    for (keys %$cond) {
      $self->throw_exception("Keys of condition should be of form 'foreign.col', not '$_'")
        if /\./ && !/^foreign\./;
    }
  }

  my %rels = %{ $self->_relationships };
  $rels{$rel} = { class => $f_source_name,
                  source => $f_source_name,
                  cond  => $cond,
                  attrs => $attrs };
  $self->_relationships(\%rels);

  return $self;

# XXX disabled. doesn't work properly currently. skip in tests.

  my $f_source = $self->schema->source($f_source_name);
  unless ($f_source) {
    $self->ensure_class_loaded($f_source_name);
    $f_source = $f_source_name->result_source;
    #my $s_class = ref($self->schema);
    #$f_source_name =~ m/^${s_class}::(.*)$/;
    #$self->schema->register_class(($1 || $f_source_name), $f_source_name);
    #$f_source = $self->schema->source($f_source_name);
  }
  return unless $f_source; # Can't test rel without f_source

  try { $self->_resolve_join($rel, 'me', {}, []) }
  catch {
    # If the resolve failed, back out and re-throw the error
    delete $rels{$rel};
    $self->_relationships(\%rels);
    $self->throw_exception("Error creating relationship $rel: $_");
  };

  1;
}

sub relationships {
  return keys %{shift->_relationships};
}

sub relationship_info {
  my ($self, $rel) = @_;
  return $self->_relationships->{$rel};
}

sub has_relationship {
  my ($self, $rel) = @_;
  return exists $self->_relationships->{$rel};
}

sub reverse_relationship_info {
  my ($self, $rel) = @_;

  my $rel_info = $self->relationship_info($rel)
    or $self->throw_exception("No such relationship '$rel'");

  my $ret = {};

  return $ret unless ((ref $rel_info->{cond}) eq 'HASH');

  my $stripped_cond = $self->__strip_relcond ($rel_info->{cond});

  my $rsrc_schema_moniker = $self->source_name
    if try { $self->schema };

  # this may be a partial schema or something else equally esoteric
  my $other_rsrc = try { $self->related_source($rel) }
    or return $ret;

  # Get all the relationships for that source that related to this source
  # whose foreign column set are our self columns on $rel and whose self
  # columns are our foreign columns on $rel
  foreach my $other_rel ($other_rsrc->relationships) {

    # only consider stuff that points back to us
    # "us" here is tricky - if we are in a schema registration, we want
    # to use the source_names, otherwise we will use the actual classes

    # the schema may be partial
    my $roundtrip_rsrc = try { $other_rsrc->related_source($other_rel) }
      or next;

    if ($rsrc_schema_moniker and try { $roundtrip_rsrc->schema } ) {
      next unless $rsrc_schema_moniker eq $roundtrip_rsrc->source_name;
    }
    else {
      next unless $self->result_class eq $roundtrip_rsrc->result_class;
    }

    my $other_rel_info = $other_rsrc->relationship_info($other_rel);

    # this can happen when we have a self-referential class
    next if $other_rel_info eq $rel_info;

    next unless ref $other_rel_info->{cond} eq 'HASH';
    my $other_stripped_cond = $self->__strip_relcond($other_rel_info->{cond});

    $ret->{$other_rel} = $other_rel_info if (
      $self->_compare_relationship_keys (
        [ keys %$stripped_cond ], [ values %$other_stripped_cond ]
      )
        and
      $self->_compare_relationship_keys (
        [ values %$stripped_cond ], [ keys %$other_stripped_cond ]
      )
    );
  }

  return $ret;
}

# all this does is removes the foreign/self prefix from a condition
sub __strip_relcond {
  +{
    map
      { map { /^ (?:foreign|self) \. (\w+) $/x } ($_, $_[1]{$_}) }
      keys %{$_[1]}
  }
}

sub compare_relationship_keys {
  carp 'compare_relationship_keys is a private method, stop calling it';
  my $self = shift;
  $self->_compare_relationship_keys (@_);
}

# Returns true if both sets of keynames are the same, false otherwise.
sub _compare_relationship_keys {
#  my ($self, $keys1, $keys2) = @_;
  return
    join ("\x00", sort @{$_[1]})
      eq
    join ("\x00", sort @{$_[2]})
  ;
}

# Returns the {from} structure used to express JOIN conditions
sub _resolve_join {
  my ($self, $join, $alias, $seen, $jpath, $parent_force_left) = @_;

  # we need a supplied one, because we do in-place modifications, no returns
  $self->throw_exception ('You must supply a seen hashref as the 3rd argument to _resolve_join')
    unless ref $seen eq 'HASH';

  $self->throw_exception ('You must supply a joinpath arrayref as the 4th argument to _resolve_join')
    unless ref $jpath eq 'ARRAY';

  $jpath = [@$jpath]; # copy

  if (not defined $join) {
    return ();
  }
  elsif (ref $join eq 'ARRAY') {
    return
      map {
        $self->_resolve_join($_, $alias, $seen, $jpath, $parent_force_left);
      } @$join;
  }
  elsif (ref $join eq 'HASH') {

    my @ret;
    for my $rel (keys %$join) {

      my $rel_info = $self->relationship_info($rel)
        or $self->throw_exception("No such relationship '$rel' on " . $self->source_name);

      my $force_left = $parent_force_left;
      $force_left ||= lc($rel_info->{attrs}{join_type}||'') eq 'left';

      # the actual seen value will be incremented by the recursion
      my $as = $self->storage->relname_to_table_alias(
        $rel, ($seen->{$rel} && $seen->{$rel} + 1)
      );

      push @ret, (
        $self->_resolve_join($rel, $alias, $seen, [@$jpath], $force_left),
        $self->related_source($rel)->_resolve_join(
          $join->{$rel}, $as, $seen, [@$jpath, { $rel => $as }], $force_left
        )
      );
    }
    return @ret;

  }
  elsif (ref $join) {
    $self->throw_exception("No idea how to resolve join reftype ".ref $join);
  }
  else {
    my $count = ++$seen->{$join};
    my $as = $self->storage->relname_to_table_alias(
      $join, ($count > 1 && $count)
    );

    my $rel_info = $self->relationship_info($join)
      or $self->throw_exception("No such relationship $join on " . $self->source_name);

    my $rel_src = $self->related_source($join);
    return [ { $as => $rel_src->from,
               -rsrc => $rel_src,
               -join_type => $parent_force_left
                  ? 'left'
                  : $rel_info->{attrs}{join_type}
                ,
               -join_path => [@$jpath, { $join => $as } ],
               -is_single => (
                  $rel_info->{attrs}{accessor}
                    &&
                  first { $rel_info->{attrs}{accessor} eq $_ } (qw/single filter/)
                ),
               -alias => $as,
               -relation_chain_depth => $seen->{-relation_chain_depth} || 0,
             },
             $self->_resolve_condition($rel_info->{cond}, $as, $alias, $join)
          ];
  }
}

sub pk_depends_on {
  carp 'pk_depends_on is a private method, stop calling it';
  my $self = shift;
  $self->_pk_depends_on (@_);
}

# Determines whether a relation is dependent on an object from this source
# having already been inserted. Takes the name of the relationship and a
# hashref of columns of the related object.
sub _pk_depends_on {
  my ($self, $relname, $rel_data) = @_;

  my $relinfo = $self->relationship_info($relname);

  # don't assume things if the relationship direction is specified
  return $relinfo->{attrs}{is_foreign_key_constraint}
    if exists ($relinfo->{attrs}{is_foreign_key_constraint});

  my $cond = $relinfo->{cond};
  return 0 unless ref($cond) eq 'HASH';

  # map { foreign.foo => 'self.bar' } to { bar => 'foo' }
  my $keyhash = { map { my $x = $_; $x =~ s/.*\.//; $x; } reverse %$cond };

  # assume anything that references our PK probably is dependent on us
  # rather than vice versa, unless the far side is (a) defined or (b)
  # auto-increment
  my $rel_source = $self->related_source($relname);

  foreach my $p ($self->primary_columns) {
    if (exists $keyhash->{$p}) {
      unless (defined($rel_data->{$keyhash->{$p}})
              || $rel_source->column_info($keyhash->{$p})
                            ->{is_auto_increment}) {
        return 0;
      }
    }
  }

  return 1;
}

sub resolve_condition {
  carp 'resolve_condition is a private method, stop calling it';
  my $self = shift;
  $self->_resolve_condition (@_);
}

our $UNRESOLVABLE_CONDITION = \ '1 = 0';

# Resolves the passed condition to a concrete query fragment and a flag
# indicating whether this is a cross-table condition. Also an optional
# list of non-triviail values (notmally conditions) returned as a part
# of a joinfree condition hash
sub _resolve_condition {
  my ($self, $cond, $as, $for, $relname) = @_;

  my $obj_rel = !!blessed $for;

  if (ref $cond eq 'CODE') {
    my $relalias = $obj_rel ? 'me' : $as;

    my ($crosstable_cond, $joinfree_cond) = $cond->({
      self_alias => $obj_rel ? $as : $for,
      foreign_alias => $relalias,
      self_resultsource => $self,
      foreign_relname => $relname || ($obj_rel ? $as : $for),
      self_rowobj => $obj_rel ? $for : undef
    });

    my $cond_cols;
    if ($joinfree_cond) {

      # FIXME sanity check until things stabilize, remove at some point
      $self->throw_exception (
        "A join-free condition returned for relationship '$relname' whithout a row-object to chain from"
      ) unless $obj_rel;

      # FIXME another sanity check
      if (
        ref $joinfree_cond ne 'HASH'
          or
        first { $_ !~ /^\Q$relalias.\E.+/ } keys %$joinfree_cond
      ) {
        $self->throw_exception (
          "The join-free condition returned for relationship '$relname' must be a hash "
         .'reference with all keys being valid columns on the related result source'
        );
      }

      # normalize
      for (values %$joinfree_cond) {
        $_ = $_->{'='} if (
          ref $_ eq 'HASH'
            and
          keys %$_ == 1
            and
          exists $_->{'='}
        );
      }

      # see which parts of the joinfree cond are conditionals
      my $relcol_list = { map { $_ => 1 } $self->related_source($relname)->columns };

      for my $c (keys %$joinfree_cond) {
        my ($colname) = $c =~ /^ (?: \Q$relalias.\E )? (.+)/x;

        unless ($relcol_list->{$colname}) {
          push @$cond_cols, $colname;
          next;
        }

        if (
          ref $joinfree_cond->{$c}
            and
          ref $joinfree_cond->{$c} ne 'SCALAR'
            and
          ref $joinfree_cond->{$c} ne 'REF'
        ) {
          push @$cond_cols, $colname;
          next;
        }
      }

      return wantarray ? ($joinfree_cond, 0, $cond_cols) : $joinfree_cond;
    }
    else {
      return wantarray ? ($crosstable_cond, 1) : $crosstable_cond;
    }
  }
  elsif (ref $cond eq 'HASH') {
    my %ret;
    foreach my $k (keys %{$cond}) {
      my $v = $cond->{$k};
      # XXX should probably check these are valid columns
      $k =~ s/^foreign\.// ||
        $self->throw_exception("Invalid rel cond key ${k}");
      $v =~ s/^self\.// ||
        $self->throw_exception("Invalid rel cond val ${v}");
      if (ref $for) { # Object
        #warn "$self $k $for $v";
        unless ($for->has_column_loaded($v)) {
          if ($for->in_storage) {
            $self->throw_exception(sprintf
              "Unable to resolve relationship '%s' from object %s: column '%s' not "
            . 'loaded from storage (or not passed to new() prior to insert()). You '
            . 'probably need to call ->discard_changes to get the server-side defaults '
            . 'from the database.',
              $as,
              $for,
              $v,
            );
          }
          return $UNRESOLVABLE_CONDITION;
        }
        $ret{$k} = $for->get_column($v);
        #$ret{$k} = $for->get_column($v) if $for->has_column_loaded($v);
        #warn %ret;
      } elsif (!defined $for) { # undef, i.e. "no object"
        $ret{$k} = undef;
      } elsif (ref $as eq 'HASH') { # reverse hashref
        $ret{$v} = $as->{$k};
      } elsif (ref $as) { # reverse object
        $ret{$v} = $as->get_column($k);
      } elsif (!defined $as) { # undef, i.e. "no reverse object"
        $ret{$v} = undef;
      } else {
        $ret{"${as}.${k}"} = { -ident => "${for}.${v}" };
      }
    }

    return wantarray
      ? ( \%ret, ($obj_rel || !defined $as || ref $as) ? 0 : 1 )
      : \%ret
    ;
  }
  elsif (ref $cond eq 'ARRAY') {
    my (@ret, $crosstable);
    for (@$cond) {
      my ($cond, $crosstab) = $self->_resolve_condition($_, $as, $for, $relname);
      push @ret, $cond;
      $crosstable ||= $crosstab;
    }
    return wantarray ? (\@ret, $crosstable) : \@ret;
  }
  else {
    $self->throw_exception ("Can't handle condition $cond for relationship '$relname' yet :(");
  }
}

# Accepts one or more relationships for the current source and returns an
# array of column names for each of those relationships. Column names are
# prefixed relative to the current source, in accordance with where they appear
# in the supplied relationships.

sub _resolve_prefetch {
  my ($self, $pre, $alias, $alias_map, $order, $collapse, $pref_path) = @_;
  $pref_path ||= [];

  if (not defined $pre) {
    return ();
  }
  elsif( ref $pre eq 'ARRAY' ) {
    return
      map { $self->_resolve_prefetch( $_, $alias, $alias_map, $order, $collapse, [ @$pref_path ] ) }
        @$pre;
  }
  elsif( ref $pre eq 'HASH' ) {
    my @ret =
    map {
      $self->_resolve_prefetch($_, $alias, $alias_map, $order, $collapse, [ @$pref_path ] ),
      $self->related_source($_)->_resolve_prefetch(
               $pre->{$_}, "${alias}.$_", $alias_map, $order, $collapse, [ @$pref_path, $_] )
    } keys %$pre;
    return @ret;
  }
  elsif( ref $pre ) {
    $self->throw_exception(
      "don't know how to resolve prefetch reftype ".ref($pre));
  }
  else {
    my $p = $alias_map;
    $p = $p->{$_} for (@$pref_path, $pre);

    $self->throw_exception (
      "Unable to resolve prefetch '$pre' - join alias map does not contain an entry for path: "
      . join (' -> ', @$pref_path, $pre)
    ) if (ref $p->{-join_aliases} ne 'ARRAY' or not @{$p->{-join_aliases}} );

    my $as = shift @{$p->{-join_aliases}};

    my $rel_info = $self->relationship_info( $pre );
    $self->throw_exception( $self->source_name . " has no such relationship '$pre'" )
      unless $rel_info;
    my $as_prefix = ($alias =~ /^.*?\.(.+)$/ ? $1.'.' : '');
    my $rel_source = $self->related_source($pre);

    if ($rel_info->{attrs}{accessor} && $rel_info->{attrs}{accessor} eq 'multi') {
      $self->throw_exception(
        "Can't prefetch has_many ${pre} (join cond too complex)")
        unless ref($rel_info->{cond}) eq 'HASH';
      my $dots = @{[$as_prefix =~ m/\./g]} + 1; # +1 to match the ".${as_prefix}"

      if (my ($fail) = grep { @{[$_ =~ m/\./g]} == $dots }
                         keys %{$collapse}) {
        my ($last) = ($fail =~ /([^\.]+)$/);
        carp (
          "Prefetching multiple has_many rels ${last} and ${pre} "
          .(length($as_prefix)
            ? "at the same level (${as_prefix}) "
            : "at top level "
          )
          . 'will explode the number of row objects retrievable via ->next or ->all. '
          . 'Use at your own risk.'
        );
      }

      #my @col = map { (/^self\.(.+)$/ ? ("${as_prefix}.$1") : ()); }
      #              values %{$rel_info->{cond}};
      $collapse->{".${as_prefix}${pre}"} = [ $rel_source->_pri_cols ];
        # action at a distance. prepending the '.' allows simpler code
        # in ResultSet->_collapse_result
      my @key = map { (/^foreign\.(.+)$/ ? ($1) : ()); }
                    keys %{$rel_info->{cond}};
      push @$order, map { "${as}.$_" } @key;

      if (my $rel_order = $rel_info->{attrs}{order_by}) {
        # this is kludgy and incomplete, I am well aware
        # but the parent method is going away entirely anyway
        # so sod it
        my $sql_maker = $self->storage->sql_maker;
        my ($orig_ql, $orig_qr) = $sql_maker->_quote_chars;
        my $sep = $sql_maker->name_sep;

        # install our own quoter, so we can catch unqualified stuff
        local $sql_maker->{quote_char} = ["\x00", "\xFF"];

        my $quoted_prefix = "\x00${as}\xFF";

        for my $chunk ( $sql_maker->_order_by_chunks ($rel_order) ) {
          my @bind;
          ($chunk, @bind) = @$chunk if ref $chunk;

          $chunk = "${quoted_prefix}${sep}${chunk}"
            unless $chunk =~ /\Q$sep/;

          $chunk =~ s/\x00/$orig_ql/g;
          $chunk =~ s/\xFF/$orig_qr/g;
          push @$order, \[$chunk, @bind];
        }
      }
    }

    return map { [ "${as}.$_", "${as_prefix}${pre}.$_", ] }
      $rel_source->columns;
  }
}

sub related_source {
  my ($self, $rel) = @_;
  if( !$self->has_relationship( $rel ) ) {
    $self->throw_exception("No such relationship '$rel' on " . $self->source_name);
  }

  # if we are not registered with a schema - just use the prototype
  # however if we do have a schema - ask for the source by name (and
  # throw in the process if all fails)
  if (my $schema = try { $self->schema }) {
    $schema->source($self->relationship_info($rel)->{source});
  }
  else {
    my $class = $self->relationship_info($rel)->{class};
    $self->ensure_class_loaded($class);
    $class->result_source_instance;
  }
}

sub related_class {
  my ($self, $rel) = @_;
  if( !$self->has_relationship( $rel ) ) {
    $self->throw_exception("No such relationship '$rel' on " . $self->source_name);
  }
  return $self->schema->class($self->relationship_info($rel)->{source});
}

sub handle {
  return DBIx::Class::ResultSourceHandle->new({
    source_moniker => $_[0]->source_name,

    # so that a detached thaw can be re-frozen
    $_[0]->{_detached_thaw}
      ? ( _detached_source  => $_[0]          )
      : ( schema            => $_[0]->schema  )
    ,
  });
}

{
  my $global_phase_destroy;

  # SpeedyCGI runs END blocks every cycle but keeps object instances
  # hence we have to disable the globaldestroy hatch, and rely on the
  # eval trap below (which appears to work, but is risky done so late)
  END { $global_phase_destroy = 1 unless $CGI::SpeedyCGI::i_am_speedy }

  sub DESTROY {
    return if $global_phase_destroy;

######
# !!! ACHTUNG !!!!
######
#
# Under no circumstances shall $_[0] be stored anywhere else (like copied to
# a lexical variable, or shifted, or anything else). Doing so will mess up
# the refcount of this particular result source, and will allow the $schema
# we are trying to save to reattach back to the source we are destroying.
# The relevant code checking refcounts is in ::Schema::DESTROY()

    # if we are not a schema instance holder - we don't matter
    return if(
      ! ref $_[0]->{schema}
        or
      isweak $_[0]->{schema}
    );

    # weaken our schema hold forcing the schema to find somewhere else to live
    # during global destruction (if we have not yet bailed out) this will throw
    # which will serve as a signal to not try doing anything else
    local $@;
    eval {
      weaken $_[0]->{schema};
      1;
    } or do {
      $global_phase_destroy = 1;
      return;
    };


    # if schema is still there reintroduce ourselves with strong refs back to us
    if ($_[0]->{schema}) {
      my $srcregs = $_[0]->{schema}->source_registrations;
      for (keys %$srcregs) {
        next unless $srcregs->{$_};
        $srcregs->{$_} = $_[0] if $srcregs->{$_} == $_[0];
      }
    }
  }
}

sub STORABLE_freeze { Storable::nfreeze($_[0]->handle) }

sub STORABLE_thaw {
  my ($self, $cloning, $ice) = @_;
  %$self = %{ (Storable::thaw($ice))->resolve };
}

sub throw_exception {
  my $self = shift;

  $self->{schema}
    ? $self->{schema}->throw_exception(@_)
    : DBIx::Class::Exception->throw(@_)
  ;
}

1;