t/73oracle_hq.t

   1 use strict;
   2 use warnings;
   3
   4 use Test::Exception;
   5 use Test::More;
   6
   7 use lib qw(t/lib);
   8 use DBIC::SqlMakerTest;
   9
  10 $ENV{NLS_SORT} = "BINARY";
  11 $ENV{NLS_COMP} = "BINARY";
  12 $ENV{NLS_LANG} = "AMERICAN";
  13
  14 plan skip_all => 'Test needs ' . DBIx::Class::Optional::Dependencies->req_missing_for ('test_rdbms_oracle')
  15   unless DBIx::Class::Optional::Dependencies->req_ok_for ('test_rdbms_oracle');
  16
  17 my ($dsn,  $user,  $pass)  = @ENV{map { "DBICTEST_ORA_${_}" }  qw/DSN USER PASS/};
  18
  19 plan skip_all => 'Set $ENV{DBICTEST_ORA_DSN}, _USER and _PASS to run this test.'
  20  unless ($dsn && $user && $pass);
  21
  22 use DBICTest::Schema::Artist;
  23 BEGIN {
  24   DBICTest::Schema::Artist->add_column('parentid');
  25
  26   DBICTest::Schema::Artist->has_many(
  27     children => 'DBICTest::Schema::Artist',
  28     { 'foreign.parentid' => 'self.artistid' }
  29   );
  30
  31   DBICTest::Schema::Artist->belongs_to(
  32     parent => 'DBICTest::Schema::Artist',
  33     { 'foreign.artistid' => 'self.parentid' }
  34   );
  35 }
  36
  37 use DBICTest::Schema;
  38
  39 my $schema = DBICTest::Schema->connect($dsn, $user, $pass);
  40
  41 note "Oracle Version: " . $schema->storage->_server_info->{dbms_version};
  42
  43 my $dbh = $schema->storage->dbh;
  44 do_creates($dbh);
  45
  46 ### test hierarchical queries
  47 {
  48   $schema->resultset('Artist')->create ({
  49     name => 'root',
  50     rank => 1,
  51     cds => [],
  52     children => [
  53       {
  54         name => 'child1',
  55         rank => 2,
  56         children => [
  57           {
  58             name => 'grandchild',
  59             rank => 3,
  60             cds => [
  61               {
  62                 title => "grandchilds's cd" ,
  63                 year => '2008',
  64                 tracks => [
  65                   {
  66                     position => 1,
  67                     title => 'Track 1 grandchild',
  68                   }
  69                 ],
  70               }
  71             ],
  72             children => [
  73               {
  74                 name => 'greatgrandchild',
  75                 rank => 3,
  76               }
  77             ],
  78           }
  79         ],
  80       },
  81       {
  82         name => 'child2',
  83         rank => 3,
  84       },
  85     ],
  86   });
  87
  88   $schema->resultset('Artist')->create({
  89     name => 'cycle-root',
  90     children => [
  91       {
  92         name => 'cycle-child1',
  93         children => [ { name => 'cycle-grandchild' } ],
  94       },
  95       {
  96         name => 'cycle-child2'
  97       },
  98     ],
  99   });
 100
 101   $schema->resultset('Artist')->find({ name => 'cycle-root' })
 102     ->update({ parentid => { -ident => 'artistid' } });
 103
 104   # select the whole tree
 105   {
 106     my $rs = $schema->resultset('Artist')->search({}, {
 107       start_with => { name => 'root' },
 108       connect_by => { parentid => { -prior => { -ident => 'artistid' } } },
 109     });
 110
 111     is_same_sql_bind (
 112       $rs->as_query,
 113       '(
 114         SELECT me.artistid, me.name, me.rank, me.charfield, me.parentid
 115           FROM artist me
 116         START WITH name = ?
 117         CONNECT BY parentid = PRIOR artistid
 118       )',
 119       [ [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'name', 'sqlt_size' => 100 }
 120             => 'root'] ],
 121     );
 122     is_deeply (
 123       [ $rs->get_column ('name')->all ],
 124       [ qw/root child1 grandchild greatgrandchild child2/ ],
 125       'got artist tree',
 126     );
 127
 128     is_same_sql_bind (
 129       $rs->count_rs->as_query,
 130       '(
 131         SELECT COUNT( * )
 132           FROM artist me
 133         START WITH name = ?
 134         CONNECT BY parentid = PRIOR artistid
 135       )',
 136       [ [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'name', 'sqlt_size' => 100 }
 137             => 'root'] ],
 138     );
 139
 140     is( $rs->count, 5, 'Connect By count ok' );
 141   }
 142
 143   # use order siblings by statement
 144   SKIP: {
 145     # http://download.oracle.com/docs/cd/A87860_01/doc/server.817/a85397/state21b.htm#2066123
 146     skip q{Oracle8i doesn't support ORDER SIBLINGS BY}, 1
 147       if $schema->storage->_server_info->{normalized_dbms_version} < 9;
 148
 149     my $rs = $schema->resultset('Artist')->search({}, {
 150       start_with => { name => 'root' },
 151       connect_by => { parentid => { -prior => { -ident =>  'artistid' } } },
 152       order_siblings_by => { -desc => 'name' },
 153     });
 154
 155     is_same_sql_bind (
 156       $rs->as_query,
 157       '(
 158         SELECT me.artistid, me.name, me.rank, me.charfield, me.parentid
 159           FROM artist me
 160         START WITH name = ?
 161         CONNECT BY parentid = PRIOR artistid
 162         ORDER SIBLINGS BY name DESC
 163       )',
 164       [ [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'name', 'sqlt_size' => 100 }
 165             => 'root'] ],
 166     );
 167
 168     is_deeply (
 169       [ $rs->get_column ('name')->all ],
 170       [ qw/root child2 child1 grandchild greatgrandchild/ ],
 171       'Order Siblings By ok',
 172     );
 173   }
 174
 175   # get the root node
 176   {
 177     my $rs = $schema->resultset('Artist')->search({ parentid => undef }, {
 178       start_with => { name => 'root' },
 179       connect_by => { parentid => { -prior => { -ident => 'artistid' } } },
 180     });
 181
 182     is_same_sql_bind (
 183       $rs->as_query,
 184       '(
 185         SELECT me.artistid, me.name, me.rank, me.charfield, me.parentid
 186           FROM artist me
 187         WHERE ( parentid IS NULL )
 188         START WITH name = ?
 189         CONNECT BY parentid = PRIOR artistid
 190       )',
 191       [ [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'name', 'sqlt_size' => 100 }
 192             => 'root'] ],
 193     );
 194
 195     is_deeply(
 196       [ $rs->get_column('name')->all ],
 197       [ 'root' ],
 198       'found root node',
 199     );
 200   }
 201
 202   # combine a connect by with a join
 203   SKIP: {
 204     # http://download.oracle.com/docs/cd/A87860_01/doc/server.817/a85397/state21b.htm#2066123
 205     skip q{Oracle8i doesn't support connect by with join}, 1
 206       if $schema->storage->_server_info->{normalized_dbms_version} < 9;
 207
 208     my $rs = $schema->resultset('Artist')->search(
 209       {'cds.title' => { -like => '%cd'} },
 210       {
 211         join => 'cds',
 212         start_with => { 'me.name' => 'root' },
 213         connect_by => { parentid => { -prior => { -ident => 'artistid' } } },
 214       }
 215     );
 216
 217     is_same_sql_bind (
 218       $rs->as_query,
 219       '(
 220         SELECT me.artistid, me.name, me.rank, me.charfield, me.parentid
 221           FROM artist me
 222           LEFT JOIN cd cds ON cds.artist = me.artistid
 223         WHERE ( cds.title LIKE ? )
 224         START WITH me.name = ?
 225         CONNECT BY parentid = PRIOR artistid
 226       )',
 227       [
 228         [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'cds.title', 'sqlt_size' => 100 }
 229             => '%cd'],
 230         [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'me.name', 'sqlt_size' => 100 }
 231             => 'root'],
 232       ],
 233     );
 234
 235     is_deeply(
 236       [ $rs->get_column('name')->all ],
 237       [ 'grandchild' ],
 238       'Connect By with a join result name ok'
 239     );
 240
 241     is_same_sql_bind (
 242       $rs->count_rs->as_query,
 243       '(
 244         SELECT COUNT( * )
 245           FROM artist me
 246           LEFT JOIN cd cds ON cds.artist = me.artistid
 247         WHERE ( cds.title LIKE ? )
 248         START WITH me.name = ?
 249         CONNECT BY parentid = PRIOR artistid
 250       )',
 251       [
 252         [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'cds.title', 'sqlt_size' => 100 }
 253             => '%cd'],
 254         [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'me.name', 'sqlt_size' => 100 }
 255             => 'root'],
 256       ],
 257     );
 258
 259     is( $rs->count, 1, 'Connect By with a join; count ok' );
 260   }
 261
 262   # combine a connect by with order_by
 263   {
 264     my $rs = $schema->resultset('Artist')->search({}, {
 265       start_with => { name => 'root' },
 266       connect_by => { parentid => { -prior => { -ident => 'artistid' } } },
 267       order_by => { -asc => [ 'LEVEL', 'name' ] },
 268     });
 269
 270     is_same_sql_bind (
 271       $rs->as_query,
 272       '(
 273         SELECT me.artistid, me.name, me.rank, me.charfield, me.parentid
 274           FROM artist me
 275         START WITH name = ?
 276         CONNECT BY parentid = PRIOR artistid
 277         ORDER BY LEVEL ASC, name ASC
 278       )',
 279       [
 280         [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'name', 'sqlt_size' => 100 }
 281             => 'root'],
 282       ],
 283     );
 284
 285
 286     # Don't use "$rs->get_column ('name')->all" they build a query arround the $rs.
 287     #   If $rs has a order by, the order by is in the subquery and this doesn't work with Oracle 8i.
 288     # TODO: write extra test and fix order by handling on Oracle 8i
 289     is_deeply (
 290       [ map { $_->[1] } $rs->cursor->all ],
 291       [ qw/root child1 child2 grandchild greatgrandchild/ ],
 292       'Connect By with a order_by - result name ok (without get_column)'
 293     );
 294
 295     SKIP: {
 296       skip q{Connect By with a order_by - result name ok (with get_column), Oracle8i doesn't support order by in a subquery},1
 297         if $schema->storage->_server_info->{normalized_dbms_version} < 9;
 298       is_deeply (
 299         [  $rs->get_column ('name')->all ],
 300         [ qw/root child1 child2 grandchild greatgrandchild/ ],
 301         'Connect By with a order_by - result name ok (with get_column)'
 302       );
 303     }
 304   }
 305
 306
 307   # limit a connect by
 308   SKIP: {
 309     skip q{Oracle8i doesn't support order by in a subquery}, 1
 310       if $schema->storage->_server_info->{normalized_dbms_version} < 9;
 311
 312     my $rs = $schema->resultset('Artist')->search({}, {
 313       start_with => { name => 'root' },
 314       connect_by => { parentid => { -prior => { -ident => 'artistid' } } },
 315       order_by => { -asc => 'name' },
 316       rows => 2,
 317     });
 318
 319     is_same_sql_bind (
 320       $rs->as_query,
 321       '(
 322         SELECT artistid, name, rank, charfield, parentid
 323           FROM (
 324             SELECT me.artistid, me.name, me.rank, me.charfield, me.parentid
 325               FROM artist me
 326             START WITH name = ?
 327             CONNECT BY parentid = PRIOR artistid
 328             ORDER BY name ASC
 329           ) me
 330         WHERE ROWNUM <= 2
 331       )',
 332       [
 333         [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'name', 'sqlt_size' => 100 }
 334             => 'root'],
 335       ],
 336     );
 337
 338     is_deeply (
 339       [ $rs->get_column ('name')->all ],
 340       [qw/child1 child2/],
 341       'LIMIT a Connect By query - correct names'
 342     );
 343
 344     is_same_sql_bind (
 345       $rs->count_rs->as_query,
 346       '(
 347         SELECT COUNT( * )
 348           FROM (
 349             SELECT artistid
 350               FROM (
 351                 SELECT me.artistid
 352                   FROM artist me
 353                 START WITH name = ?
 354                 CONNECT BY parentid = PRIOR artistid
 355               ) me
 356             WHERE ROWNUM <= 2
 357           ) me
 358       )',
 359       [
 360         [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'name', 'sqlt_size' => 100 }
 361             => 'root'],
 362       ],
 363     );
 364
 365     is( $rs->count, 2, 'Connect By; LIMIT count ok' );
 366   }
 367
 368   # combine a connect_by with group_by and having
 369   # add some bindvals to make sure things still work
 370   {
 371     my $rs = $schema->resultset('Artist')->search({}, {
 372       select => \[ 'COUNT(rank) + ?', [ __cbind => 3 ] ],
 373       as => 'cnt',
 374       start_with => { name => 'root' },
 375       connect_by => { parentid => { -prior => { -ident => 'artistid' } } },
 376       group_by => \[ 'rank + ? ', [ __gbind =>  1] ],
 377       having => \[ 'count(rank) < ?', [ cnt => 2 ] ],
 378     });
 379
 380     is_same_sql_bind (
 381       $rs->as_query,
 382       '(
 383         SELECT COUNT(rank) + ?
 384           FROM artist me
 385         START WITH name = ?
 386         CONNECT BY parentid = PRIOR artistid
 387         GROUP BY( rank + ? ) HAVING count(rank) < ?
 388       )',
 389       [
 390         [ { dbic_colname => '__cbind' }
 391             => 3 ],
 392         [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'name', 'sqlt_size' => 100 }
 393             => 'root'],
 394         [ { dbic_colname => '__gbind' }
 395             => 1 ],
 396         [ { dbic_colname => 'cnt' }
 397             => 2 ],
 398       ],
 399     );
 400
 401     is_deeply (
 402       [ $rs->get_column ('cnt')->all ],
 403       [4, 4],
 404       'Group By a Connect By query - correct values'
 405     );
 406   }
 407
 408   # select the whole cycle tree without nocylce
 409   {
 410     my $rs = $schema->resultset('Artist')->search({}, {
 411       start_with => { name => 'cycle-root' },
 412       connect_by => { parentid => { -prior => { -ident => 'artistid' } } },
 413     });
 414
 415     # ORA-01436:  CONNECT BY loop in user data
 416     throws_ok { $rs->get_column ('name')->all } qr/ORA-01436/,
 417       "connect by initify loop detection without nocycle";
 418   }
 419
 420   # select the whole cycle tree with nocylce
 421   SKIP: {
 422     # http://download.oracle.com/docs/cd/A87860_01/doc/server.817/a85397/expressi.htm#1023748
 423     skip q{Oracle8i doesn't support connect by nocycle}, 1
 424       if $schema->storage->_server_info->{normalized_dbms_version} < 9;
 425
 426     my $rs = $schema->resultset('Artist')->search({}, {
 427       start_with => { name => 'cycle-root' },
 428       '+select'  => \ 'CONNECT_BY_ISCYCLE',
 429       '+as'      => [ 'connector' ],
 430       connect_by_nocycle => { parentid => { -prior => { -ident => 'artistid' } } },
 431     });
 432
 433     is_same_sql_bind (
 434       $rs->as_query,
 435       '(
 436         SELECT me.artistid, me.name, me.rank, me.charfield, me.parentid, CONNECT_BY_ISCYCLE
 437           FROM artist me
 438         START WITH name = ?
 439         CONNECT BY NOCYCLE parentid = PRIOR artistid
 440       )',
 441       [
 442         [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'name', 'sqlt_size' => 100 }
 443             => 'cycle-root'],
 444       ],
 445     );
 446     is_deeply (
 447       [ $rs->get_column ('name')->all ],
 448       [ qw/cycle-root cycle-child1 cycle-grandchild cycle-child2/ ],
 449       'got artist tree with nocycle (name)',
 450     );
 451     is_deeply (
 452       [ $rs->get_column ('connector')->all ],
 453       [ qw/1 0 0 0/ ],
 454       'got artist tree with nocycle (CONNECT_BY_ISCYCLE)',
 455     );
 456
 457     is_same_sql_bind (
 458       $rs->count_rs->as_query,
 459       '(
 460         SELECT COUNT( * )
 461           FROM artist me
 462         START WITH name = ?
 463         CONNECT BY NOCYCLE parentid = PRIOR artistid
 464       )',
 465       [
 466         [ { 'sqlt_datatype' => 'varchar', 'dbic_colname' => 'name', 'sqlt_size' => 100 }
 467             => 'cycle-root'],
 468       ],
 469     );
 470
 471     is( $rs->count, 4, 'Connect By Nocycle count ok' );
 472   }
 473 }
 474
 475 done_testing;
 476
 477 sub do_creates {
 478   my $dbh = shift;
 479
 480   eval {
 481     $dbh->do("DROP SEQUENCE artist_autoinc_seq");
 482     $dbh->do("DROP SEQUENCE artist_pk_seq");
 483     $dbh->do("DROP SEQUENCE cd_seq");
 484     $dbh->do("DROP SEQUENCE track_seq");
 485     $dbh->do("DROP TABLE artist");
 486     $dbh->do("DROP TABLE track");
 487     $dbh->do("DROP TABLE cd");
 488   };
 489
 490   $dbh->do("CREATE SEQUENCE artist_pk_seq START WITH 1 MAXVALUE 999999 MINVALUE 0");
 491   $dbh->do("CREATE SEQUENCE cd_seq START WITH 1 MAXVALUE 999999 MINVALUE 0");
 492   $dbh->do("CREATE SEQUENCE track_seq START WITH 1 MAXVALUE 999999 MINVALUE 0");
 493
 494   $dbh->do("CREATE TABLE artist (artistid NUMBER(12), parentid NUMBER(12), name VARCHAR(255), autoinc_col NUMBER(12), rank NUMBER(38), charfield VARCHAR2(10))");
 495   $dbh->do("ALTER TABLE artist ADD (CONSTRAINT artist_pk PRIMARY KEY (artistid))");
 496
 497   $dbh->do("CREATE TABLE cd (cdid NUMBER(12), artist NUMBER(12), title VARCHAR(255), year VARCHAR(4), genreid NUMBER(12), single_track NUMBER(12))");
 498   $dbh->do("ALTER TABLE cd ADD (CONSTRAINT cd_pk PRIMARY KEY (cdid))");
 499
 500   $dbh->do("CREATE TABLE track (trackid NUMBER(12), cd NUMBER(12) REFERENCES cd(cdid) DEFERRABLE, position NUMBER(12), title VARCHAR(255), last_updated_on DATE, last_updated_at DATE, small_dt DATE)");
 501   $dbh->do("ALTER TABLE track ADD (CONSTRAINT track_pk PRIMARY KEY (trackid))");
 502
 503   $dbh->do(qq{
 504     CREATE OR REPLACE TRIGGER artist_insert_trg_pk
 505     BEFORE INSERT ON artist
 506     FOR EACH ROW
 507       BEGIN
 508         IF :new.artistid IS NULL THEN
 509           SELECT artist_pk_seq.nextval
 510           INTO :new.artistid
 511           FROM DUAL;
 512         END IF;
 513       END;
 514   });
 515   $dbh->do(qq{
 516     CREATE OR REPLACE TRIGGER cd_insert_trg
 517     BEFORE INSERT OR UPDATE ON cd
 518     FOR EACH ROW
 519
 520       DECLARE
 521       tmpVar NUMBER;
 522
 523       BEGIN
 524         tmpVar := 0;
 525
 526         IF :new.cdid IS NULL THEN
 527           SELECT cd_seq.nextval
 528           INTO tmpVar
 529           FROM dual;
 530
 531           :new.cdid := tmpVar;
 532         END IF;
 533       END;
 534   });
 535   $dbh->do(qq{
 536     CREATE OR REPLACE TRIGGER track_insert_trg
 537     BEFORE INSERT ON track
 538     FOR EACH ROW
 539       BEGIN
 540         IF :new.trackid IS NULL THEN
 541           SELECT track_seq.nextval
 542           INTO :new.trackid
 543           FROM DUAL;
 544         END IF;
 545       END;
 546   });
 547 }
 548
 549 # clean up our mess
 550 END {
 551   eval {
 552     my $dbh = $schema->storage->dbh;
 553     $dbh->do("DROP SEQUENCE artist_pk_seq");
 554     $dbh->do("DROP SEQUENCE cd_seq");
 555     $dbh->do("DROP SEQUENCE track_seq");
 556     $dbh->do("DROP TABLE artist");
 557     $dbh->do("DROP TABLE track");
 558     $dbh->do("DROP TABLE cd");
 559   };
 560 }