Rose::DB::Pg - PostgreSQL driver class for Rose::DB.

  use Rose::DB;

  Rose::DB->register_db(
    domain   => 'development',
    type     => 'main',
    driver   => 'Pg',
    database => 'dev_db',
    host     => 'localhost',
    username => 'devuser',
    password => 'mysecret',
    server_time_zone => 'UTC',
    european_dates   => 1,
  );

  Rose::DB->default_domain('development');
  Rose::DB->default_type('main');
  ...

  $db = Rose::DB->new; # $db is really a Rose::DB::Pg-derived object
  ...

package Rose::DB::Pg;

use strict;

use DateTime::Infinite;
use DateTime::Format::Pg;
use SQL::ReservedWords::PostgreSQL();

use Rose::DB;

our $VERSION = '0.786'; # overshot version number, freeze until caught up

our $Debug = 0;

#
# Object data
#

use Rose::Object::MakeMethods::Generic
(
  'scalar' =>
  [
    qw(sslmode service options)
  ],
);

#
# Object methods
#

sub build_dsn
{
  my($self_or_class, %args) = @_;

  my %info;

  $info{'dbname'} = $args{'db'} || $args{'database'};

  @info{qw(host port options service sslmode)} =
    @args{qw(host port options service sslmode)};

  return
    "dbi:Pg:" . 
    join(';', map { "$_=$info{$_}" } grep { defined $info{$_} }
              qw(dbname host port options service sslmode));
}

sub dbi_driver { 'Pg' }

sub init_date_handler
{
  my($self) = shift;

  my $parent_class = ref($self)->parent_class;
  my $european_dates   = "${parent_class}::european_dates";
  my $server_time_zone = "${parent_class}::server_time_zone";

  no strict 'refs';
  my $parser = 
    DateTime::Format::Pg->new(
      ($self->$european_dates() ? (european => 1) : ()),
      ($self->$server_time_zone() ? 
        (server_tz => $self->$server_time_zone()) : ()));

  return $parser;
}

sub default_implicit_schema { 'public' }
sub likes_lowercase_table_names    { 1 }
sub likes_lowercase_schema_names   { 1 }
sub likes_lowercase_catalog_names  { 1 }
sub likes_lowercase_sequence_names { 1 }

sub supports_multi_column_count_distinct  { 0 }
sub supports_arbitrary_defaults_on_insert { 1 }
sub supports_select_from_subselect        { 1 }

sub pg_enable_utf8 { shift->dbh_attribute_boolean('pg_enable_utf8', @_) }

sub supports_schema { 1 }

sub max_column_name_length { 63 }
sub max_column_alias_length { 63 }

sub last_insertid_from_sth
{
  #my($self, $sth, $obj) = @_;

  # PostgreSQL demands that the primary key column not be in the insert
  # statement at all in order for it to auto-generate a value.  The
  # insert SQL will need to be modified to make this work for
  # Rose::DB::Object...
  #if($DBD::Pg::VERSION >= 1.40)
  #{
  #  my $meta = $obj->meta;
  #  return $self->dbh->last_insert_id(undef, $meta->select_schema, $meta->table, undef);
  #}

  return undef;
}

sub format_select_lock
{
  my($self, $class, $lock, $tables_list) = @_;

  $lock = { type => $lock }  unless(ref $lock);

  $lock->{'type'} ||= 'for update'  if($lock->{'for_update'});

  my %types =
  (
    'for update' => 'FOR UPDATE',
    'shared'     => 'FOR SHARE',
  );

  my $sql = $types{$lock->{'type'}}
    or Carp::croak "Invalid lock type: $lock->{'type'}";

  my @tables;

  if(my $on = $lock->{'on'})
  {
    @tables = map { $self->table_sql_from_lock_on_value($class, $_, $tables_list) } @$on;
  }
  elsif(my $lock_tables = $lock->{'tables'})
  {
    my %map;

    if($tables_list)
    {
      my $tn = 1;

      foreach my $table (@$tables_list)
      {
        (my $table_key = $table) =~ s/^(["']?)[^.]+\1\.//;
        $map{$table_key} = 't' . $tn++;
      }
    }

    @tables = map
      {
        ref $_ eq 'SCALAR' ? $$_ :
          $self->auto_quote_table_name(defined $map{$_} ? $map{$_} : $_)
      }
      @$lock_tables;
  }

  if(@tables)
  {
    $sql .= ' OF ' . join(', ', @tables);
  }

  $sql .= ' NOWAIT'  if($lock->{'nowait'});

  return $sql;
}

sub parse_datetime
{
  my($self) = shift;

  unless(ref $_[0])
  {
    no warnings 'uninitialized';
    return DateTime::Infinite::Past->new   if($_[0] eq '-infinity');
    return DateTime::Infinite::Future->new if($_[0] eq 'infinity');
  }

  my $method = ref($self)->parent_class . '::parse_datetime';

  no strict 'refs';
  $self->$method(@_);
}

sub parse_timestamp
{
  my($self) = shift;

  unless(ref $_[0])
  {
    no warnings 'uninitialized';
    return DateTime::Infinite::Past->new   if($_[0] eq '-infinity');
    return DateTime::Infinite::Future->new if($_[0] eq 'infinity');
  }

  my $method = ref($self)->parent_class . '::parse_timestamp';

  no strict 'refs';
  $self->$method(@_);
}

sub parse_timestamp_with_time_zone
{
  my($self, $value) = @_;

  unless(ref $value)
  {
    no warnings 'uninitialized';
    return DateTime::Infinite::Past->new   if($value eq '-infinity');
    return DateTime::Infinite::Future->new if($value eq 'infinity');
  }

  my $method = ref($self)->parent_class . '::parse_timestamp_with_time_zone';

  no strict 'refs';
  shift->$method(@_);
}

sub validate_date_keyword
{
  no warnings;
  $_[1] =~ /^(?:(?:now|timeofday)(?:\(\))?|(?:current_(?:date|time(?:stamp)?)
        |localtime(?:stamp)?)(?:\(\d*\))?|epoch|today|tomorrow|yesterday|)$/xi ||
    ($_[0]->keyword_function_calls && $_[1] =~ /^\w+\(.*\)$/);
}

sub validate_time_keyword
{
  no warnings;
  $_[1] =~ /^(?:(?:now|timeofday)(?:\(\))?|(?:current_(?:date|time(?:stamp)?)
        |localtime(?:stamp)?)(?:\(\d*\))?|allballs)$/xi ||
    ($_[0]->keyword_function_calls && $_[1] =~ /^\w+\(.*\)$/);
}

sub validate_timestamp_keyword
{
  no warnings;
  $_[1] =~ /^(?:(?:now|timeofday)(?:\(\))?|(?:current_(?:date|time(?:stamp)?)
        |localtime(?:stamp)?)(?:\(\d*\))?|-?infinity|epoch|today|tomorrow|yesterday|allballs)$/xi ||
    ($_[0]->keyword_function_calls && $_[1] =~ /^\w+\(.*\)$/);

}

*validate_datetime_keyword = \&validate_timestamp_keyword;

sub server_time_zone
{
  my($self) = shift;

  $self->{'date_handler'} = undef  if(@_);

  my $method = ref($self)->parent_class . '::server_time_zone';

  no strict 'refs';
  $self->$method(@_);
}

sub european_dates
{
  my($self) = shift;

  $self->{'date_handler'} = undef  if(@_);

  my $method = ref($self)->parent_class . '::european_dates';

  no strict 'refs';
  $self->$method(@_);
}

sub parse_array
{
  my($self) = shift;

  return $_[0]  if(ref $_[0]);
  return [ @_ ] if(@_ > 1);

  my $val = $_[0];

  return undef  unless(defined $val);

  $val =~ s/^ (?:\[.+\]=)? \{ (.*) \} $/$1/sx;

  my @array;

  while($val =~ s/(?:"((?:[^"\\]+|\\.)*)"|([^",]+))(?:,|$)//)
  {
    my($item) = map { $_ eq 'NULL' ? undef : $_ } (defined $1 ? $1 : $2);
    $item =~ s{\\(.)}{$1}g  if(defined $item);
    push(@array, $item);
  }

  return \@array;
}

sub format_array
{
  my($self) = shift;

  return undef  unless(ref $_[0] || defined $_[0]);

  my @array = (ref $_[0]) ? @{$_[0]} : @_;

  return '{' . join(',', map 
  {
    if(!defined $_)
    {
      'NULL'
    }
    elsif(/^[-+]?\d+(?:\.\d*)?$/)
    {
      $_
    }
    else
    {
      s/\\/\\\\/g; 
      s/"/\\"/g;
      qq("$_") 
    }
  } @array) . '}';
}

sub parse_interval
{
  my($self, $value, $end_of_month_mode) = @_;

  if(!defined $value || UNIVERSAL::isa($value, 'DateTime::Duration') || 
     $self->validate_interval_keyword($value) || 
     ($self->keyword_function_calls && $value =~ /^\w+\(.*\)$/))
  {
    return $value;
  }

  my($dt_duration, $error);

  TRY:
  {
    local $@;
    eval { $dt_duration = $self->date_handler->parse_interval($value) };
    $error = $@;
  }

  my $method = ref($self)->parent_class . '::parse_interval';

  no strict 'refs';
  return $self->$method($value, $end_of_month_mode)  if($error);

  if(defined $end_of_month_mode && $dt_duration)
  {
    # XXX: There is no mutator for end_of_month_mode, so I'm being evil
    # XXX: and setting it directly.  Blah.
    $dt_duration->{'end_of_month'} = $end_of_month_mode;
  }

  return $dt_duration;
}

BEGIN
{
  require DateTime::Format::Pg;

  # Handle DateTime::Format::Pg bug
  # http://rt.cpan.org/Public/Bug/Display.html?id=18487  
  if($DateTime::Format::Pg::VERSION < 0.11)
  {
    *format_interval = sub
    {
      my($self, $dur) = @_;
      return $dur  if(!defined $dur || $self->validate_interval_keyword($dur) ||
        ($self->keyword_function_calls && $dur =~ /^\w+\(.*\)$/));
      my $val = $self->date_handler->format_interval($dur);

      $val =~ s/(\S+e\S+) seconds/sprintf('%f seconds', $1)/e;
      return $val;
    };
  }
  else
  {
    *format_interval = sub
    {
      my($self, $dur) = @_;
      return $dur  if(!defined $dur || $self->validate_interval_keyword($dur) ||
        ($self->keyword_function_calls && $dur =~ /^\w+\(.*\)$/));
      return $self->date_handler->format_interval($dur);
    };
  }
}

sub next_value_in_sequence
{
  my($self, $sequence_name) = @_;

  my $dbh = $self->dbh or return undef;

  my($value, $error);

  TRY:
  {
    local $@;

    eval
    {
      local $dbh->{'PrintError'} = 0;
      local $dbh->{'RaiseError'} = 1;
      my $sth = $dbh->prepare(qq(SELECT nextval(?)));
      $sth->execute($sequence_name);
      $value = ${$sth->fetchrow_arrayref}[0];
    };

    $error = $@;
  }

  if($error)
  {
    $self->error("Could not get the next value in the sequence '$sequence_name' - $error");
    return undef;
  }

  return $value;
}

sub current_value_in_sequence
{
  my($self, $sequence_name) = @_;

  my $dbh = $self->dbh or return undef;

  my($value, $error);

  TRY:
  {
    local $@;

    eval
    {
      local $dbh->{'PrintError'} = 0;
      local $dbh->{'RaiseError'} = 1;
      my $name = $dbh->quote_identifier($sequence_name);
      my $sth = $dbh->prepare(qq(SELECT last_value FROM $name));
      $sth->execute;
      $value = ${$sth->fetchrow_arrayref}[0];
    };

    $error = $@;
  }

  if($error)
  {
    $self->error("Could not get the current value in the sequence '$sequence_name' - $error");
    return undef;
  }

  return $value;
}

sub sequence_exists { defined shift->current_value_in_sequence(@_) ? 1 : 0 }

sub use_auto_sequence_name { 1 }

sub auto_sequence_name
{
  my($self, %args) = @_;

  my $table = $args{'table'};
  Carp::croak "Missing table argument"  unless(defined $table);

  my $column = $args{'column'};
  Carp::croak "Missing column argument"  unless(defined $column);

  return lc "${table}_${column}_seq";
}

*is_reserved_word = \&SQL::ReservedWords::PostgreSQL::is_reserved;

#
# DBI introspection
#

sub refine_dbi_column_info
{
  my($self, $col_info, $meta) = @_;

  # Save default value
  my $default = $col_info->{'COLUMN_DEF'};

  my $method = ref($self)->parent_class . '::refine_dbi_column_info';

  no strict 'refs';
  $self->$method($col_info);


  if(defined $default)
  {
    # Set sequence name key, if present
    if($default =~ /^nextval\(\(?'((?:''|[^']+))'::\w+/)
    {
      $col_info->{'rdbo_default_value_sequence_name'} = 
        $self->likes_lowercase_sequence_names ? lc $1 : $1;

      if($meta)
      {
        my $seq = $col_info->{'rdbo_default_value_sequence_name'};

        my $implicit_schema = $self->default_implicit_schema;

        # Strip off default implicit schema unless a schema is explicitly 
        # specified in the RDBO metadata object.
        if(defined $seq && defined $implicit_schema && !defined $meta->schema)
        {
          $seq =~ s/^$implicit_schema\.//;
        }

        $col_info->{'rdbo_default_value_sequence_name'} = $self->unquote_column_name($seq);

        # Pg returns serial columns as integer or bigint
        if($col_info->{'TYPE_NAME'} eq 'integer' ||
           $col_info->{'TYPE_NAME'} eq 'bigint')
        {
          my $db = $meta->db;

          my $auto_seq =
            $db->auto_sequence_name(table  => $meta->table,
                                    column => $col_info->{'COLUMN_NAME'});

          # Use schema prefix on auto-generated name if necessary
          if($seq =~ /^[^.]+\./)
          {
            my $schema = $meta->select_schema($db);
            $auto_seq = "$schema.$auto_seq"  if($schema);
          }

          no warnings 'uninitialized';
          if(lc $seq eq lc $auto_seq)
          {
            $col_info->{'TYPE_NAME'} =
              $col_info->{'TYPE_NAME'} eq 'integer' ? 'serial' : 'bigserial';
          }
        }
      }
    }
    elsif($default =~ /^NULL::[\w ]+$/)
    {
      # RT 64331: https://rt.cpan.org/Ticket/Display.html?id=64331
      $col_info->{'COLUMN_DEF'} = undef;
    }
  }

  my $type_name = $col_info->{'TYPE_NAME'};

  # Pg has some odd/different names for types.  Convert them to standard forms.
  if($type_name eq 'character varying')
  {
    $col_info->{'TYPE_NAME'} = 'varchar';
  }
  elsif($type_name eq 'bit')
  {
    $col_info->{'TYPE_NAME'} = 'bits';
  }
  elsif($type_name eq 'real')
  {
    $col_info->{'TYPE_NAME'} = 'float';
  }
  elsif($type_name eq 'time without time zone')
  {
    $col_info->{'TYPE_NAME'} = 'time';
    $col_info->{'pg_type'} =~ /^time(?:\((\d+)\))? without time zone$/i;
    $col_info->{'TIME_SCALE'} = $1 || 0;
  }
  elsif($type_name eq 'double precision')
  {
    $col_info->{'COLUMN_SIZE'} = undef;
  }
  elsif($type_name eq 'money')
  {
    $col_info->{'COLUMN_SIZE'} = undef;
  }

  # Pg does not populate COLUMN_SIZE correctly for bit fields, so
  # we have to extract the number of bits from pg_type.
  if($col_info->{'pg_type'} =~ /^bit\((\d+)\)$/)
  {
    $col_info->{'COLUMN_SIZE'} = $1;
  }

  # Extract precision and scale from numeric types
  if($col_info->{'pg_type'} =~ /^numeric/i)
  {
    no warnings 'uninitialized';

    if($col_info->{'COLUMN_SIZE'} =~ /^(\d+),(\d+)$/)
    {
      $col_info->{'COLUMN_SIZE'}    = $1;
      $col_info->{'DECIMAL_DIGITS'} = $2;
    }
    elsif($col_info->{'pg_type'} =~ /^numeric\((\d+),(\d+)\)$/i)
    {
      $col_info->{'COLUMN_SIZE'}    = $2;
      $col_info->{'DECIMAL_DIGITS'} = $1;
    }
  }

  # Treat custom types that look like enums as enums
  if(ref $col_info->{'pg_enum_values'} && @{$col_info->{'pg_enum_values'}})
  {
    $col_info->{'TYPE_NAME'} = 'enum';
    $col_info->{'RDBO_ENUM_VALUES'} = $col_info->{'pg_enum_values'};
    $col_info->{'RDBO_DB_TYPE'} = $col_info->{'pg_type'};
  }

  # We currently treat all arrays the same, regardless of what they are 
  # arrays of: integer, character, float, etc.  So we covert TYPE_NAMEs
  # like 'integer[]' into 'array'
  if($col_info->{'TYPE_NAME'} =~ /^\w.*\[\]$/)
  {
    $col_info->{'TYPE_NAME'} = 'array';
  }

  return;
}

sub parse_dbi_column_info_default 
{
  my($self, $string, $col_info) = @_;

  no warnings 'uninitialized';
  local $_ = $string;

  my $pg_vers = $self->dbh->{'pg_server_version'};

  # Example: q(B'00101'::"bit")
  if(/^B'([01]+)'::(?:bit|"bit")$/ && $col_info->{'TYPE_NAME'} eq 'bit')
  {
    return $1;
  }
  # Example: 922337203685::bigint
  elsif(/^(.+)::"?bigint"?$/i && $col_info->{'TYPE_NAME'} eq 'bigint')
  {
    return $1;
  }
  # TODO: http://rt.cpan.org/Ticket/Display.html?id=35462
  # Example: '{foo,"\\"bar,",baz}'::text[]
  # ...
  # Example: 'value'::character varying
  # Example: ('now'::text)::timestamp(0)
  elsif(/^\(*'(.*)'::.+$/)
  {
    my $default = $1;

    # Single quotes are backslash-escaped, but PostgreSQL 8.1 and
    # later uses doubled quotes '' instead.  Strangely, I see
    # doubled quotes in 8.0.x as well...
    if($pg_vers >= 80000 && index($default, q('')) > 0)
    {
      $default =~ s/''/'/g;
    }
    elsif($pg_vers < 80100 && index($default, q(\')) > 0)
    {
      $default = $1;
      $default =~ s/\\'/'/g;
    }

    return $default;
  }
  # Handle sequence-based defaults elsewhere
  elsif(/^nextval\(/)
  {
    return undef;
  }

  return $string;
}

sub list_tables
{
  my($self, %args) = @_;

  my $types = $args{'include_views'} ? "'TABLE','VIEW'" : 'TABLE';
  my @tables;

  my $schema = $self->schema;
  $schema = $self->default_implicit_schema  unless(defined $schema);

  my $error;

  TRY:
  {
    local $@;

    eval
    {
      my $dbh = $self->dbh or die $self->error;

      local $dbh->{'RaiseError'} = 1;
      local $dbh->{'FetchHashKeyName'} = 'NAME';

      my $sth = $dbh->table_info($self->catalog, $schema, '', $types,
                                 { noprefix => 1, pg_noprefix => 1 });

      $sth->execute;

      while(my $table_info = $sth->fetchrow_hashref)
      {
        push(@tables, $self->unquote_table_name($table_info->{'TABLE_NAME'}));
      }
    };

    $error = $@;
  }

  if($error)
  {
    Carp::croak "Could not list tables from ", $self->dsn, " - $error";
  }

  return wantarray ? @tables : \@tables;
}

# sub list_tables
# {
#   my($self) = shift;
# 
#   my @tables;
# 
#   my $schema = $self->schema;
#   $schema = $db->default_implicit_schema  unless(defined $schema);
#     
#   if($DBD::Pg::VERSION >= 1.31) 
#   {
#     @tables = $self->dbh->tables($self->catalog, $schema, '', 'TABLE',
#                               { noprefix => 1, pg_noprefix => 1 });
#     }
#     else 
#     {
#       @tables = $dbh->tables;
#     }
#   }
# 
#   return wantarray ? @tables : \@tables;
# }

1;

__END__