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