lib/DBIx/Class/Storage/DBI/Sybase.pm

   1 package DBIx::Class::Storage::DBI::Sybase;
   2
   3 use strict;
   4 use warnings;
   5
   6 use base qw/
   7     DBIx::Class::Storage::DBI::Sybase::Common
   8     DBIx::Class::Storage::DBI::AutoCast
   9 /;
  10 use mro 'c3';
  11 use Carp::Clan qw/^DBIx::Class/;
  12 use List::Util();
  13 use Sub::Name();
  14 use Data::Dumper::Concise();
  15
  16 __PACKAGE__->mk_group_accessors('simple' =>
  17     qw/_identity _blob_log_on_update _writer_storage _is_extra_storage
  18        _bulk_storage _is_bulk_storage _began_bulk_work
  19        _bulk_disabled_due_to_coderef_connect_info_warned
  20        _identity_method/
  21 );
  22
  23 my @also_proxy_to_extra_storages = qw/
  24   connect_call_set_auto_cast auto_cast connect_call_blob_setup
  25   connect_call_datetime_setup
  26
  27   disconnect _connect_info _sql_maker _sql_maker_opts disable_sth_caching
  28   auto_savepoint unsafe cursor_class debug debugobj schema
  29 /;
  30
  31 =head1 NAME
  32
  33 DBIx::Class::Storage::DBI::Sybase - Sybase support for DBIx::Class
  34
  35 =head1 SYNOPSIS
  36
  37 This subclass supports L<DBD::Sybase> for real Sybase databases.  If you are
  38 using an MSSQL database via L<DBD::Sybase>, your storage will be reblessed to
  39 L<DBIx::Class::Storage::DBI::Sybase::Microsoft_SQL_Server>.
  40
  41 =head1 DESCRIPTION
  42
  43 If your version of Sybase does not support placeholders, then your storage
  44 will be reblessed to L<DBIx::Class::Storage::DBI::Sybase::NoBindVars>. You can
  45 also enable that driver explicitly, see the documentation for more details.
  46
  47 With this driver there is unfortunately no way to get the C<last_insert_id>
  48 without doing a C<SELECT MAX(col)>. This is done safely in a transaction
  49 (locking the table.) See L</INSERTS WITH PLACEHOLDERS>.
  50
  51 A recommended L<DBIx::Class::Storage::DBI/connect_info> setting:
  52
  53   on_connect_call => [['datetime_setup'], ['blob_setup', log_on_update => 0]]
  54
  55 =head1 METHODS
  56
  57 =cut
  58
  59 sub _rebless {
  60   my $self = shift;
  61
  62   if (ref($self) eq 'DBIx::Class::Storage::DBI::Sybase') {
  63     my $dbtype = eval {
  64       @{$self->_get_dbh->selectrow_arrayref(qq{sp_server_info \@attribute_id=1})}[2]
  65     } || '';
  66
  67     my $exception = $@;
  68     $dbtype =~ s/\W/_/gi;
  69     my $subclass = "DBIx::Class::Storage::DBI::Sybase::${dbtype}";
  70
  71     if (!$exception && $dbtype && $self->load_optional_class($subclass)) {
  72       bless $self, $subclass;
  73       $self->_rebless;
  74     } else { # real Sybase
  75       my $no_bind_vars = 'DBIx::Class::Storage::DBI::Sybase::NoBindVars';
  76
  77       if ($self->using_freetds) {
  78         carp <<'EOF' unless $ENV{DBIC_SYBASE_FREETDS_NOWARN};
  79
  80 You are using FreeTDS with Sybase.
  81
  82 We will do our best to support this configuration, but please consider this
  83 support experimental.
  84
  85 TEXT/IMAGE columns will definitely not work.
  86
  87 You are encouraged to recompile DBD::Sybase with the Sybase Open Client libraries
  88 instead.
  89
  90 See perldoc DBIx::Class::Storage::DBI::Sybase for more details.
  91
  92 To turn off this warning set the DBIC_SYBASE_FREETDS_NOWARN environment
  93 variable.
  94 EOF
  95         if (not $self->_typeless_placeholders_supported) {
  96           if ($self->_placeholders_supported) {
  97             $self->auto_cast(1);
  98           } else {
  99             $self->ensure_class_loaded($no_bind_vars);
 100             bless $self, $no_bind_vars;
 101             $self->_rebless;
 102           }
 103         }
 104       }
 105       elsif (not $self->_get_dbh->{syb_dynamic_supported}) {
 106         # not necessarily FreeTDS, but no placeholders nevertheless
 107         $self->ensure_class_loaded($no_bind_vars);
 108         bless $self, $no_bind_vars;
 109         $self->_rebless;
 110       } elsif (not $self->_typeless_placeholders_supported) {
 111         # this is highly unlikely, but we check just in case
 112         $self->auto_cast(1);
 113       }
 114     }
 115   }
 116 }
 117
 118 sub _init {
 119   my $self = shift;
 120   $self->_set_max_connect(256);
 121
 122   # based on LongReadLen in connect_info
 123   $self->set_textsize if $self->using_freetds;
 124
 125 # create storage for insert/(update blob) transactions,
 126 # unless this is that storage
 127   return if $self->_is_extra_storage;
 128
 129   my $writer_storage = (ref $self)->new;
 130
 131   $writer_storage->_is_extra_storage(1);
 132   $writer_storage->connect_info($self->connect_info);
 133   $writer_storage->auto_cast($self->auto_cast);
 134
 135   $self->_writer_storage($writer_storage);
 136
 137 # create a bulk storage unless connect_info is a coderef
 138   return
 139     if (Scalar::Util::reftype($self->_dbi_connect_info->[0])||'') eq 'CODE';
 140
 141   my $bulk_storage = (ref $self)->new;
 142
 143   $bulk_storage->_is_extra_storage(1);
 144   $bulk_storage->_is_bulk_storage(1); # for special ->disconnect acrobatics
 145   $bulk_storage->connect_info($self->connect_info);
 146
 147 # this is why
 148   $bulk_storage->_dbi_connect_info->[0] .= ';bulkLogin=1';
 149
 150   $self->_bulk_storage($bulk_storage);
 151 }
 152
 153 for my $method (@also_proxy_to_extra_storages) {
 154   no strict 'refs';
 155   no warnings 'redefine';
 156
 157   my $replaced = __PACKAGE__->can($method);
 158
 159   *{$method} = Sub::Name::subname $method => sub {
 160     my $self = shift;
 161     $self->_writer_storage->$replaced(@_) if $self->_writer_storage;
 162     $self->_bulk_storage->$replaced(@_)   if $self->_bulk_storage;
 163     return $self->$replaced(@_);
 164   };
 165 }
 166
 167 sub disconnect {
 168   my $self = shift;
 169
 170 # Even though we call $sth->finish for uses off the bulk API, there's still an
 171 # "active statement" warning on disconnect, which we throw away here.
 172 # This is due to the bug described in insert_bulk.
 173 # Currently a noop because 'prepare' is used instead of 'prepare_cached'.
 174   local $SIG{__WARN__} = sub {
 175     warn $_[0] unless $_[0] =~ /active statement/i;
 176   } if $self->_is_bulk_storage;
 177
 178 # so that next transaction gets a dbh
 179   $self->_began_bulk_work(0) if $self->_is_bulk_storage;
 180
 181   $self->next::method;
 182 }
 183
 184 # Make sure we have CHAINED mode turned on if AutoCommit is off in non-FreeTDS
 185 # DBD::Sybase (since we don't know how DBD::Sybase was compiled.) If however
 186 # we're using FreeTDS, CHAINED mode turns on an implicit transaction which we
 187 # only want when AutoCommit is off.
 188 sub _populate_dbh {
 189   my $self = shift;
 190
 191   $self->next::method(@_);
 192
 193   if ($self->_is_bulk_storage) {
 194 # this should be cleared on every reconnect
 195     $self->_began_bulk_work(0);
 196     return;
 197   }
 198
 199   if (not $self->using_freetds) {
 200     $self->_dbh->{syb_chained_txn} = 1;
 201   } else {
 202     if ($self->_dbh_autocommit) {
 203       $self->_dbh->do('SET CHAINED OFF');
 204     } else {
 205       $self->_dbh->do('SET CHAINED ON');
 206     }
 207   }
 208 }
 209
 210 =head2 connect_call_blob_setup
 211
 212 Used as:
 213
 214   on_connect_call => [ [ 'blob_setup', log_on_update => 0 ] ]
 215
 216 Does C<< $dbh->{syb_binary_images} = 1; >> to return C<IMAGE> data as raw binary
 217 instead of as a hex string.
 218
 219 Recommended.
 220
 221 Also sets the C<log_on_update> value for blob write operations. The default is
 222 C<1>, but C<0> is better if your database is configured for it.
 223
 224 See
 225 L<DBD::Sybase/Handling_IMAGE/TEXT_data_with_syb_ct_get_data()/syb_ct_send_data()>.
 226
 227 =cut
 228
 229 sub connect_call_blob_setup {
 230   my $self = shift;
 231   my %args = @_;
 232   my $dbh = $self->_dbh;
 233   $dbh->{syb_binary_images} = 1;
 234
 235   $self->_blob_log_on_update($args{log_on_update})
 236     if exists $args{log_on_update};
 237 }
 238
 239 sub _is_lob_type {
 240   my $self = shift;
 241   my $type = shift;
 242   $type && $type =~ /(?:text|image|lob|bytea|binary|memo)/i;
 243 }
 244
 245 sub _is_lob_column {
 246   my ($self, $source, $column) = @_;
 247
 248   return $self->_is_lob_type($source->column_info($column)->{data_type});
 249 }
 250
 251 sub _prep_for_execute {
 252   my $self = shift;
 253   my ($op, $extra_bind, $ident, $args) = @_;
 254
 255   my ($sql, $bind) = $self->next::method (@_);
 256
 257   my $table = Scalar::Util::blessed($ident) ? $ident->from : $ident;
 258
 259   my $bind_info = $self->_resolve_column_info(
 260     $ident, [map $_->[0], @{$bind}]
 261   );
 262   my $bound_identity_col = List::Util::first
 263     { $bind_info->{$_}{is_auto_increment} }
 264     (keys %$bind_info)
 265   ;
 266   my $identity_col = Scalar::Util::blessed($ident) &&
 267     List::Util::first
 268     { $ident->column_info($_)->{is_auto_increment} }
 269     $ident->columns
 270   ;
 271
 272   if (($op eq 'insert' && $bound_identity_col) ||
 273       ($op eq 'update' && exists $args->[0]{$identity_col})) {
 274     $sql = join ("\n",
 275       $self->_set_table_identity_sql($op => $table, 'on'),
 276       $sql,
 277       $self->_set_table_identity_sql($op => $table, 'off'),
 278     );
 279   }
 280
 281   if ($op eq 'insert' && (not $bound_identity_col) && $identity_col &&
 282       (not $self->{insert_bulk})) {
 283     $sql =
 284       "$sql\n" .
 285       $self->_fetch_identity_sql($ident, $identity_col);
 286   }
 287
 288   return ($sql, $bind);
 289 }
 290
 291 sub _set_table_identity_sql {
 292   my ($self, $op, $table, $on_off) = @_;
 293
 294   return sprintf 'SET IDENTITY_%s %s %s',
 295     uc($op), $self->sql_maker->_quote($table), uc($on_off);
 296 }
 297
 298 # Stolen from SQLT, with some modifications. This is a makeshift
 299 # solution before a sane type-mapping library is available, thus
 300 # the 'our' for easy overrides.
 301 our %TYPE_MAPPING  = (
 302     number    => 'numeric',
 303     money     => 'money',
 304     varchar   => 'varchar',
 305     varchar2  => 'varchar',
 306     timestamp => 'datetime',
 307     text      => 'varchar',
 308     real      => 'double precision',
 309     comment   => 'text',
 310     bit       => 'bit',
 311     tinyint   => 'smallint',
 312     float     => 'double precision',
 313     serial    => 'numeric',
 314     bigserial => 'numeric',
 315     boolean   => 'varchar',
 316     long      => 'varchar',
 317 );
 318
 319 sub _native_data_type {
 320   my ($self, $type) = @_;
 321
 322   $type = lc $type;
 323   $type =~ s/\s* identity//x;
 324
 325   return uc($TYPE_MAPPING{$type} || $type);
 326 }
 327
 328 sub _fetch_identity_sql {
 329   my ($self, $source, $col) = @_;
 330
 331   return sprintf ("SELECT MAX(%s) FROM %s",
 332     map { $self->sql_maker->_quote ($_) } ($col, $source->from)
 333   );
 334 }
 335
 336 sub _execute {
 337   my $self = shift;
 338   my ($op) = @_;
 339
 340   my ($rv, $sth, @bind) = $self->dbh_do($self->can('_dbh_execute'), @_);
 341
 342   if ($op eq 'insert') {
 343     $self->_identity($sth->fetchrow_array);
 344     $sth->finish;
 345   }
 346
 347   return wantarray ? ($rv, $sth, @bind) : $rv;
 348 }
 349
 350 sub last_insert_id { shift->_identity }
 351
 352 # handles TEXT/IMAGE and transaction for last_insert_id
 353 sub insert {
 354   my $self = shift;
 355   my ($source, $to_insert) = @_;
 356
 357   my $identity_col = (List::Util::first
 358     { $source->column_info($_)->{is_auto_increment} }
 359     $source->columns) || '';
 360
 361   # check for empty insert
 362   # INSERT INTO foo DEFAULT VALUES -- does not work with Sybase
 363   # try to insert explicit 'DEFAULT's instead (except for identity)
 364   if (not %$to_insert) {
 365     for my $col ($source->columns) {
 366       next if $col eq $identity_col;
 367       $to_insert->{$col} = \'DEFAULT';
 368     }
 369   }
 370
 371   my $blob_cols = $self->_remove_blob_cols($source, $to_insert);
 372
 373   # do we need the horrific SELECT MAX(COL) hack?
 374   my $dumb_last_insert_id =
 375        $identity_col
 376     && (not exists $to_insert->{$identity_col})
 377     && ($self->_identity_method||'') ne '@@IDENTITY';
 378
 379   my $next = $self->next::can;
 380
 381   # we are already in a transaction, or there are no blobs
 382   # and we don't need the PK - just (try to) do it
 383   if ($self->{transaction_depth}
 384         || (!$blob_cols && !$dumb_last_insert_id)
 385   ) {
 386     return $self->_insert (
 387       $next, $source, $to_insert, $blob_cols, $identity_col
 388     );
 389   }
 390
 391   # otherwise use the _writer_storage to do the insert+transaction on another
 392   # connection
 393   my $guard = $self->_writer_storage->txn_scope_guard;
 394
 395   my $updated_cols = $self->_writer_storage->_insert (
 396     $next, $source, $to_insert, $blob_cols, $identity_col
 397   );
 398
 399   $self->_identity($self->_writer_storage->_identity);
 400
 401   $guard->commit;
 402
 403   return $updated_cols;
 404 }
 405
 406 sub _insert {
 407   my ($self, $next, $source, $to_insert, $blob_cols, $identity_col) = @_;
 408
 409   my $updated_cols = $self->$next ($source, $to_insert);
 410
 411   my $final_row = {
 412     ($identity_col ?
 413       ($identity_col => $self->last_insert_id($source, $identity_col)) : ()),
 414     %$to_insert,
 415     %$updated_cols,
 416   };
 417
 418   $self->_insert_blobs ($source, $blob_cols, $final_row) if $blob_cols;
 419
 420   return $updated_cols;
 421 }
 422
 423 sub update {
 424   my $self = shift;
 425   my ($source, $fields, $where, @rest) = @_;
 426
 427   my $wantarray = wantarray;
 428
 429   my $blob_cols = $self->_remove_blob_cols($source, $fields);
 430
 431   my $table = $source->name;
 432
 433   my $identity_col = List::Util::first
 434     { $source->column_info($_)->{is_auto_increment} }
 435     $source->columns;
 436
 437   my $is_identity_update = $identity_col && defined $fields->{$identity_col};
 438
 439   return $self->next::method(@_) unless $blob_cols;
 440
 441 # If there are any blobs in $where, Sybase will return a descriptive error
 442 # message.
 443 # XXX blobs can still be used with a LIKE query, and this should be handled.
 444
 445 # update+blob update(s) done atomically on separate connection
 446   $self = $self->_writer_storage;
 447
 448   my $guard = $self->txn_scope_guard;
 449
 450 # First update the blob columns to be updated to '' (taken from $fields, where
 451 # it is originally put by _remove_blob_cols .)
 452   my %blobs_to_empty = map { ($_ => delete $fields->{$_}) } keys %$blob_cols;
 453
 454 # We can't only update NULL blobs, because blobs cannot be in the WHERE clause.
 455
 456   $self->next::method($source, \%blobs_to_empty, $where, @rest);
 457
 458 # Now update the blobs before the other columns in case the update of other
 459 # columns makes the search condition invalid.
 460   $self->_update_blobs($source, $blob_cols, $where);
 461
 462   my @res;
 463   if (%$fields) {
 464     if ($wantarray) {
 465       @res    = $self->next::method(@_);
 466     }
 467     elsif (defined $wantarray) {
 468       $res[0] = $self->next::method(@_);
 469     }
 470     else {
 471       $self->next::method(@_);
 472     }
 473   }
 474
 475   $guard->commit;
 476
 477   return $wantarray ? @res : $res[0];
 478 }
 479
 480 sub insert_bulk {
 481   my $self = shift;
 482   my ($source, $cols, $data) = @_;
 483
 484   my $identity_col = List::Util::first
 485     { $source->column_info($_)->{is_auto_increment} }
 486     $source->columns;
 487
 488   my $is_identity_insert = (List::Util::first
 489     { $_ eq $identity_col }
 490     @{$cols}
 491   ) ? 1 : 0;
 492
 493   my @source_columns = $source->columns;
 494
 495   my $use_bulk_api =
 496     $self->_bulk_storage &&
 497     $self->_get_dbh->{syb_has_blk};
 498
 499   if ((not $use_bulk_api) &&
 500       (Scalar::Util::reftype($self->_dbi_connect_info->[0])||'') eq 'CODE' &&
 501       (not $self->_bulk_disabled_due_to_coderef_connect_info_warned)) {
 502     carp <<'EOF';
 503 Bulk API support disabled due to use of a CODEREF connect_info. Reverting to
 504 regular array inserts.
 505 EOF
 506     $self->_bulk_disabled_due_to_coderef_connect_info_warned(1);
 507   }
 508
 509   if (not $use_bulk_api) {
 510     my $blob_cols = $self->_remove_blob_cols_array($source, $cols, $data);
 511
 512 # _execute_array uses a txn anyway, but it ends too early in case we need to
 513 # select max(col) to get the identity for inserting blobs.
 514     ($self, my $guard) = $self->{transaction_depth} == 0 ?
 515       ($self->_writer_storage, $self->_writer_storage->txn_scope_guard)
 516       :
 517       ($self, undef);
 518
 519     local $self->{insert_bulk} = 1;
 520
 521     $self->next::method(@_);
 522
 523     if ($blob_cols) {
 524       if ($is_identity_insert) {
 525         $self->_insert_blobs_array ($source, $blob_cols, $cols, $data);
 526       }
 527       else {
 528         my @cols_with_identities = (@$cols, $identity_col);
 529
 530         ## calculate identities
 531         # XXX This assumes identities always increase by 1, which may or may not
 532         # be true.
 533         my ($last_identity) =
 534           $self->_dbh->selectrow_array (
 535             $self->_fetch_identity_sql($source, $identity_col)
 536           );
 537         my @identities = (($last_identity - @$data + 1) .. $last_identity);
 538
 539         my @data_with_identities = map [@$_, shift @identities], @$data;
 540
 541         $self->_insert_blobs_array (
 542           $source, $blob_cols, \@cols_with_identities, \@data_with_identities
 543         );
 544       }
 545     }
 546
 547     $guard->commit if $guard;
 548
 549     return;
 550   }
 551
 552 # otherwise, use the bulk API
 553
 554 # rearrange @$data so that columns are in database order
 555   my %orig_idx;
 556   @orig_idx{@$cols} = 0..$#$cols;
 557
 558   my %new_idx;
 559   @new_idx{@source_columns} = 0..$#source_columns;
 560
 561   my @new_data;
 562   for my $datum (@$data) {
 563     my $new_datum = [];
 564     for my $col (@source_columns) {
 565 # identity data will be 'undef' if not $is_identity_insert
 566 # columns with defaults will also be 'undef'
 567       $new_datum->[ $new_idx{$col} ] =
 568         exists $orig_idx{$col} ? $datum->[ $orig_idx{$col} ] : undef;
 569     }
 570     push @new_data, $new_datum;
 571   }
 572
 573 # bcp identity index is 1-based
 574   my $identity_idx = exists $new_idx{$identity_col} ?
 575     $new_idx{$identity_col} + 1 : 0;
 576
 577 ## Set a client-side conversion error handler, straight from DBD::Sybase docs.
 578 # This ignores any data conversion errors detected by the client side libs, as
 579 # they are usually harmless.
 580   my $orig_cslib_cb = DBD::Sybase::set_cslib_cb(
 581     Sub::Name::subname insert_bulk => sub {
 582       my ($layer, $origin, $severity, $errno, $errmsg, $osmsg, $blkmsg) = @_;
 583
 584       return 1 if $errno == 36;
 585
 586       carp
 587         "Layer: $layer, Origin: $origin, Severity: $severity, Error: $errno" .
 588         ($errmsg ? "\n$errmsg" : '') .
 589         ($osmsg  ? "\n$osmsg"  : '')  .
 590         ($blkmsg ? "\n$blkmsg" : '');
 591
 592       return 0;
 593   });
 594
 595   eval {
 596     my $bulk = $self->_bulk_storage;
 597
 598     my $guard = $bulk->txn_scope_guard;
 599
 600 ## XXX get this to work instead of our own $sth
 601 ## will require SQLA or *Hacks changes for ordered columns
 602 #    $bulk->next::method($source, \@source_columns, \@new_data, {
 603 #      syb_bcp_attribs => {
 604 #        identity_flag   => $is_identity_insert,
 605 #        identity_column => $identity_idx,
 606 #      }
 607 #    });
 608     my $sql = 'INSERT INTO ' .
 609       $bulk->sql_maker->_quote($source->name) . ' (' .
 610 # colname list is ignored for BCP, but does no harm
 611       (join ', ', map $bulk->sql_maker->_quote($_), @source_columns) . ') '.
 612       ' VALUES ('.  (join ', ', ('?') x @source_columns) . ')';
 613
 614 ## XXX there's a bug in the DBD::Sybase bulk support that makes $sth->finish for
 615 ## a prepare_cached statement ineffective. Replace with ->sth when fixed, or
 616 ## better yet the version above. Should be fixed in DBD::Sybase .
 617     my $sth = $bulk->_get_dbh->prepare($sql,
 618 #      'insert', # op
 619       {
 620         syb_bcp_attribs => {
 621           identity_flag   => $is_identity_insert,
 622           identity_column => $identity_idx,
 623         }
 624       }
 625     );
 626
 627     my @bind = do {
 628       my $idx = 0;
 629       map [ $_, $idx++ ], @source_columns;
 630     };
 631
 632     $self->_execute_array(
 633       $source, $sth, \@bind, \@source_columns, \@new_data, sub {
 634         $guard->commit
 635       }
 636     );
 637
 638     $bulk->_query_end($sql);
 639   };
 640
 641   my $exception = $@;
 642   DBD::Sybase::set_cslib_cb($orig_cslib_cb);
 643
 644   if ($exception =~ /-Y option/) {
 645     carp <<"EOF";
 646
 647 Sybase bulk API operation failed due to character set incompatibility, reverting
 648 to regular array inserts:
 649
 650 *** Try unsetting the LANG environment variable.
 651
 652 $exception
 653 EOF
 654     $self->_bulk_storage(undef);
 655     unshift @_, $self;
 656     goto \&insert_bulk;
 657   }
 658   elsif ($exception) {
 659 # rollback makes the bulkLogin connection unusable
 660     $self->_bulk_storage->disconnect;
 661     $self->throw_exception($exception);
 662   }
 663 }
 664
 665 sub _dbh_execute_array {
 666   my ($self, $sth, $tuple_status, $cb) = @_;
 667
 668   my $rv = $self->next::method($sth, $tuple_status);
 669   $cb->() if $cb;
 670
 671   return $rv;
 672 }
 673
 674 # Make sure blobs are not bound as placeholders, and return any non-empty ones
 675 # as a hash.
 676 sub _remove_blob_cols {
 677   my ($self, $source, $fields) = @_;
 678
 679   my %blob_cols;
 680
 681   for my $col (keys %$fields) {
 682     if ($self->_is_lob_column($source, $col)) {
 683       my $blob_val = delete $fields->{$col};
 684       if (not defined $blob_val) {
 685         $fields->{$col} = \'NULL';
 686       }
 687       else {
 688         $fields->{$col} = \"''";
 689         $blob_cols{$col} = $blob_val unless $blob_val eq '';
 690       }
 691     }
 692   }
 693
 694   return %blob_cols ? \%blob_cols : undef;
 695 }
 696
 697 # same for insert_bulk
 698 sub _remove_blob_cols_array {
 699   my ($self, $source, $cols, $data) = @_;
 700
 701   my @blob_cols;
 702
 703   for my $i (0..$#$cols) {
 704     my $col = $cols->[$i];
 705
 706     if ($self->_is_lob_column($source, $col)) {
 707       for my $j (0..$#$data) {
 708         my $blob_val = delete $data->[$j][$i];
 709         if (not defined $blob_val) {
 710           $data->[$j][$i] = \'NULL';
 711         }
 712         else {
 713           $data->[$j][$i] = \"''";
 714           $blob_cols[$j][$i] = $blob_val
 715             unless $blob_val eq '';
 716         }
 717       }
 718     }
 719   }
 720
 721   return @blob_cols ? \@blob_cols : undef;
 722 }
 723
 724 sub _update_blobs {
 725   my ($self, $source, $blob_cols, $where) = @_;
 726
 727   my (@primary_cols) = $source->primary_columns;
 728
 729   $self->throw_exception('Cannot update TEXT/IMAGE column(s) without a primary key')
 730     unless @primary_cols;
 731
 732 # check if we're updating a single row by PK
 733   my $pk_cols_in_where = 0;
 734   for my $col (@primary_cols) {
 735     $pk_cols_in_where++ if defined $where->{$col};
 736   }
 737   my @rows;
 738
 739   if ($pk_cols_in_where == @primary_cols) {
 740     my %row_to_update;
 741     @row_to_update{@primary_cols} = @{$where}{@primary_cols};
 742     @rows = \%row_to_update;
 743   } else {
 744     my $cursor = $self->select ($source, \@primary_cols, $where, {});
 745     @rows = map {
 746       my %row; @row{@primary_cols} = @$_; \%row
 747     } $cursor->all;
 748   }
 749
 750   for my $row (@rows) {
 751     $self->_insert_blobs($source, $blob_cols, $row);
 752   }
 753 }
 754
 755 sub _insert_blobs {
 756   my ($self, $source, $blob_cols, $row) = @_;
 757   my $dbh = $self->_get_dbh;
 758
 759   my $table = $source->name;
 760
 761   my %row = %$row;
 762   my (@primary_cols) = $source->primary_columns;
 763
 764   $self->throw_exception('Cannot update TEXT/IMAGE column(s) without a primary key')
 765     unless @primary_cols;
 766
 767   $self->throw_exception('Cannot update TEXT/IMAGE column(s) without primary key values')
 768     if ((grep { defined $row{$_} } @primary_cols) != @primary_cols);
 769
 770   for my $col (keys %$blob_cols) {
 771     my $blob = $blob_cols->{$col};
 772
 773     my %where = map { ($_, $row{$_}) } @primary_cols;
 774
 775     my $cursor = $self->select ($source, [$col], \%where, {});
 776     $cursor->next;
 777     my $sth = $cursor->sth;
 778
 779     if (not $sth) {
 780
 781       $self->throw_exception(
 782           "Could not find row in table '$table' for blob update:\n"
 783         . Data::Dumper::Concise::Dumper (\%where)
 784       );
 785     }
 786
 787     eval {
 788       do {
 789         $sth->func('CS_GET', 1, 'ct_data_info') or die $sth->errstr;
 790       } while $sth->fetch;
 791
 792       $sth->func('ct_prepare_send') or die $sth->errstr;
 793
 794       my $log_on_update = $self->_blob_log_on_update;
 795       $log_on_update    = 1 if not defined $log_on_update;
 796
 797       $sth->func('CS_SET', 1, {
 798         total_txtlen => length($blob),
 799         log_on_update => $log_on_update
 800       }, 'ct_data_info') or die $sth->errstr;
 801
 802       $sth->func($blob, length($blob), 'ct_send_data') or die $sth->errstr;
 803
 804       $sth->func('ct_finish_send') or die $sth->errstr;
 805     };
 806     my $exception = $@;
 807     $sth->finish if $sth;
 808     if ($exception) {
 809       if ($self->using_freetds) {
 810         $self->throw_exception (
 811           'TEXT/IMAGE operation failed, probably because you are using FreeTDS: '
 812           . $exception
 813         );
 814       } else {
 815         $self->throw_exception($exception);
 816       }
 817     }
 818   }
 819 }
 820
 821 sub _insert_blobs_array {
 822   my ($self, $source, $blob_cols, $cols, $data) = @_;
 823
 824   for my $i (0..$#$data) {
 825     my $datum = $data->[$i];
 826
 827     my %row;
 828     @row{ @$cols } = @$datum;
 829
 830     my %blob_vals;
 831     for my $j (0..$#$cols) {
 832       if (exists $blob_cols->[$i][$j]) {
 833         $blob_vals{ $cols->[$j] } = $blob_cols->[$i][$j];
 834       }
 835     }
 836
 837     $self->_insert_blobs ($source, \%blob_vals, \%row);
 838   }
 839 }
 840
 841 =head2 connect_call_datetime_setup
 842
 843 Used as:
 844
 845   on_connect_call => 'datetime_setup'
 846
 847 In L<DBIx::Class::Storage::DBI/connect_info> to set:
 848
 849   $dbh->syb_date_fmt('ISO_strict'); # output fmt: 2004-08-21T14:36:48.080Z
 850   $dbh->do('set dateformat mdy');   # input fmt:  08/13/1979 18:08:55.080
 851
 852 On connection for use with L<DBIx::Class::InflateColumn::DateTime>, using
 853 L<DateTime::Format::Sybase>, which you will need to install.
 854
 855 This works for both C<DATETIME> and C<SMALLDATETIME> columns, although
 856 C<SMALLDATETIME> columns only have minute precision.
 857
 858 =cut
 859
 860 {
 861   my $old_dbd_warned = 0;
 862
 863   sub connect_call_datetime_setup {
 864     my $self = shift;
 865     my $dbh = $self->_get_dbh;
 866
 867     if ($dbh->can('syb_date_fmt')) {
 868       # amazingly, this works with FreeTDS
 869       $dbh->syb_date_fmt('ISO_strict');
 870     } elsif (not $old_dbd_warned) {
 871       carp "Your DBD::Sybase is too old to support ".
 872       "DBIx::Class::InflateColumn::DateTime, please upgrade!";
 873       $old_dbd_warned = 1;
 874     }
 875
 876     $dbh->do('SET DATEFORMAT mdy');
 877
 878     1;
 879   }
 880 }
 881
 882 sub datetime_parser_type { "DateTime::Format::Sybase" }
 883
 884 # ->begin_work and such have no effect with FreeTDS but we run them anyway to
 885 # let the DBD keep any state it needs to.
 886 #
 887 # If they ever do start working, the extra statements will do no harm (because
 888 # Sybase supports nested transactions.)
 889
 890 sub _dbh_begin_work {
 891   my $self = shift;
 892
 893 # bulkLogin=1 connections are always in a transaction, and can only call BEGIN
 894 # TRAN once. However, we need to make sure there's a $dbh.
 895   return if $self->_is_bulk_storage && $self->_dbh && $self->_began_bulk_work;
 896
 897   $self->next::method(@_);
 898
 899   if ($self->using_freetds) {
 900     $self->_get_dbh->do('BEGIN TRAN');
 901   }
 902
 903   $self->_began_bulk_work(1) if $self->_is_bulk_storage;
 904 }
 905
 906 sub _dbh_commit {
 907   my $self = shift;
 908   if ($self->using_freetds) {
 909     $self->_dbh->do('COMMIT');
 910   }
 911   return $self->next::method(@_);
 912 }
 913
 914 sub _dbh_rollback {
 915   my $self = shift;
 916   if ($self->using_freetds) {
 917     $self->_dbh->do('ROLLBACK');
 918   }
 919   return $self->next::method(@_);
 920 }
 921
 922 # savepoint support using ASE syntax
 923
 924 sub _svp_begin {
 925   my ($self, $name) = @_;
 926
 927   $self->_get_dbh->do("SAVE TRANSACTION $name");
 928 }
 929
 930 # A new SAVE TRANSACTION with the same name releases the previous one.
 931 sub _svp_release { 1 }
 932
 933 sub _svp_rollback {
 934   my ($self, $name) = @_;
 935
 936   $self->_get_dbh->do("ROLLBACK TRANSACTION $name");
 937 }
 938
 939 1;
 940
 941 =head1 Schema::Loader Support
 942
 943 There is an experimental branch of L<DBIx::Class::Schema::Loader> that will
 944 allow you to dump a schema from most (if not all) versions of Sybase.
 945
 946 It is available via subversion from:
 947
 948   http://dev.catalyst.perl.org/repos/bast/branches/DBIx-Class-Schema-Loader/current/
 949
 950 =head1 FreeTDS
 951
 952 This driver supports L<DBD::Sybase> compiled against FreeTDS
 953 (L<http://www.freetds.org/>) to the best of our ability, however it is
 954 recommended that you recompile L<DBD::Sybase> against the Sybase Open Client
 955 libraries. They are a part of the Sybase ASE distribution:
 956
 957 The Open Client FAQ is here:
 958 L<http://www.isug.com/Sybase_FAQ/ASE/section7.html>.
 959
 960 Sybase ASE for Linux (which comes with the Open Client libraries) may be
 961 downloaded here: L<http://response.sybase.com/forms/ASE_Linux_Download>.
 962
 963 To see if you're using FreeTDS check C<< $schema->storage->using_freetds >>, or run:
 964
 965   perl -MDBI -le 'my $dbh = DBI->connect($dsn, $user, $pass); print $dbh->{syb_oc_version}'
 966
 967 Some versions of the libraries involved will not support placeholders, in which
 968 case the storage will be reblessed to
 969 L<DBIx::Class::Storage::DBI::Sybase::NoBindVars>.
 970
 971 In some configurations, placeholders will work but will throw implicit type
 972 conversion errors for anything that's not expecting a string. In such a case,
 973 the C<auto_cast> option from L<DBIx::Class::Storage::DBI::AutoCast> is
 974 automatically set, which you may enable on connection with
 975 L<DBIx::Class::Storage::DBI::AutoCast/connect_call_set_auto_cast>. The type info
 976 for the C<CAST>s is taken from the L<DBIx::Class::ResultSource/data_type>
 977 definitions in your Result classes, and are mapped to a Sybase type (if it isn't
 978 already) using a mapping based on L<SQL::Translator>.
 979
 980 In other configurations, placeholers will work just as they do with the Sybase
 981 Open Client libraries.
 982
 983 Inserts or updates of TEXT/IMAGE columns will B<NOT> work with FreeTDS.
 984
 985 =head1 INSERTS WITH PLACEHOLDERS
 986
 987 With placeholders enabled, inserts are done in a transaction so that there are
 988 no concurrency issues with getting the inserted identity value using
 989 C<SELECT MAX(col)>, which is the only way to get the C<IDENTITY> value in this
 990 mode.
 991
 992 In addition, they are done on a separate connection so that it's possible to
 993 have active cursors when doing an insert.
 994
 995 When using C<DBIx::Class::Storage::DBI::Sybase::NoBindVars> transactions are
 996 disabled, as there are no concurrency issues with C<SELECT @@IDENTITY> as it's a
 997 session variable.
 998
 999 =head1 TRANSACTIONS
1000
1001 Due to limitations of the TDS protocol, L<DBD::Sybase>, or both; you cannot
1002 begin a transaction while there are active cursors; nor can you use multiple
1003 active cursors within a transaction. An active cursor is, for example, a
1004 L<ResultSet|DBIx::Class::ResultSet> that has been executed using C<next> or
1005 C<first> but has not been exhausted or L<reset|DBIx::Class::ResultSet/reset>.
1006
1007 For example, this will not work:
1008
1009   $schema->txn_do(sub {
1010     my $rs = $schema->resultset('Book');
1011     while (my $row = $rs->next) {
1012       $schema->resultset('MetaData')->create({
1013         book_id => $row->id,
1014         ...
1015       });
1016     }
1017   });
1018
1019 This won't either:
1020
1021   my $first_row = $large_rs->first;
1022   $schema->txn_do(sub { ... });
1023
1024 Transactions done for inserts in C<AutoCommit> mode when placeholders are in use
1025 are not affected, as they are done on an extra database handle.
1026
1027 Some workarounds:
1028
1029 =over 4
1030
1031 =item * use L<DBIx::Class::Storage::DBI::Replicated>
1032
1033 =item * L<connect|DBIx::Class::Schema/connect> another L<Schema|DBIx::Class::Schema>
1034
1035 =item * load the data from your cursor with L<DBIx::Class::ResultSet/all>
1036
1037 =back
1038
1039 =head1 MAXIMUM CONNECTIONS
1040
1041 The TDS protocol makes separate connections to the server for active statements
1042 in the background. By default the number of such connections is limited to 25,
1043 on both the client side and the server side.
1044
1045 This is a bit too low for a complex L<DBIx::Class> application, so on connection
1046 the client side setting is set to C<256> (see L<DBD::Sybase/maxConnect>.) You
1047 can override it to whatever setting you like in the DSN.
1048
1049 See
1050 L<http://infocenter.sybase.com/help/index.jsp?topic=/com.sybase.help.ase_15.0.sag1/html/sag1/sag1272.htm>
1051 for information on changing the setting on the server side.
1052
1053 =head1 DATES
1054
1055 See L</connect_call_datetime_setup> to setup date formats
1056 for L<DBIx::Class::InflateColumn::DateTime>.
1057
1058 =head1 TEXT/IMAGE COLUMNS
1059
1060 L<DBD::Sybase> compiled with FreeTDS will B<NOT> allow you to insert or update
1061 C<TEXT/IMAGE> columns.
1062
1063 Setting C<< $dbh->{LongReadLen} >> will also not work with FreeTDS use either:
1064
1065   $schema->storage->dbh->do("SET TEXTSIZE $bytes");
1066
1067 or
1068
1069   $schema->storage->set_textsize($bytes);
1070
1071 instead.
1072
1073 However, the C<LongReadLen> you pass in
1074 L<DBIx::Class::Storage::DBI/connect_info> is used to execute the equivalent
1075 C<SET TEXTSIZE> command on connection.
1076
1077 See L</connect_call_blob_setup> for a L<DBIx::Class::Storage::DBI/connect_info>
1078 setting you need to work with C<IMAGE> columns.
1079
1080 =head1 BULK API
1081
1082 The experimental L<DBD::Sybase> Bulk API support is used for
1083 L<populate|DBIx::Class::ResultSet/populate> in B<void> context, in a transaction
1084 on a separate connection.
1085
1086 To use this feature effectively, use a large number of rows for each
1087 L<populate|DBIx::Class::ResultSet/populate> call, eg.:
1088
1089   while (my $rows = $data_source->get_100_rows()) {
1090     $rs->populate($rows);
1091   }
1092
1093 B<NOTE:> the L<add_columns|DBIx::Class::ResultSource/add_columns>
1094 calls in your C<Result> classes B<must> list columns in database order for this
1095 to work. Also, you may have to unset the C<LANG> environment variable before
1096 loading your app, if it doesn't match the character set of your database.
1097
1098 When inserting IMAGE columns using this method, you'll need to use
1099 L</connect_call_blob_setup> as well.
1100
1101 =head1 TODO
1102
1103 =over
1104
1105 =item *
1106
1107 Transitions to AutoCommit=0 (starting a transaction) mode by exhausting
1108 any active cursors, using eager cursors.
1109
1110 =item *
1111
1112 Real limits and limited counts using stored procedures deployed on startup.
1113
1114 =item *
1115
1116 Adaptive Server Anywhere (ASA) support, with possible SQLA::Limit support.
1117
1118 =item *
1119
1120 Blob update with a LIKE query on a blob, without invalidating the WHERE condition.
1121
1122 =item *
1123
1124 bulk_insert using prepare_cached (see comments.)
1125
1126 =back
1127
1128 =head1 AUTHOR
1129
1130 See L<DBIx::Class/CONTRIBUTORS>.
1131
1132 =head1 LICENSE
1133
1134 You may distribute this code under the same terms as Perl itself.
1135
1136 =cut
1137 # vim:sts=2 sw=2: