Oracle AQ Propagation

Refer to the following Metalink documents for troubleshooting issues;
368237.1
382994.1
118884.1
203225.1
1233675.1
Bug 8467005

Ensure DST is the same on both databases, 977512.1
select version from v$timezone_file;

Execute the following on the DESTINATION schema first;
conn destination_schema/password@destination_db

variable p_Schema varchar2(30)
exec :p_Schema := sys_context('userenv','current_schema');

exec dbms_aqadm.stop_queue(:p_Schema||'.AQ_EMS_QUEUE');
exec dbms_aqadm.drop_queue(:p_Schema||'.AQ_EMS_QUEUE');
begin
  dbms_aqadm.drop_queue_table(queue_table=>:p_Schema||'.T_AQ_EMS_QUEUE'
                             ,force=>true);
end;
/
/* Ensure payload type is created prior to queue creation */
create or replace type ty_Msg as object (id number
                                        ,msg varchar2(100));
/
begin
  dbms_aqadm.create_queue_table(queue_table=>:p_Schema||'.T_AQ_EMS_QUEUE'
                               ,queue_payload_type=>'TY_MSG'
                               ,multiple_consumers=>true);
end;
/

begin
  dbms_aqadm.create_queue(queue_name =>:p_Schema||'.AQ_EMS_QUEUE'
                         ,queue_table=>:p_Schema||'.T_AQ_EMS_QUEUE');
end;
/
begin
  dbms_aqadm.start_queue (queue_name => :p_Schema||'.AQ_EMS_QUEUE');
end;
/
set serveroutput on
declare
  already_subscriber  exception;

begin
  dbms_aqadm.add_subscriber(queue_name=>:p_Schema|| '.AQ_EMS_QUEUE'
                           ,subscriber=> sys.aq$_agent('SUB2'
                                                      ,null
                                                      ,null));
  dbms_output.put_line('Subscriber sucessfully set');
end;
/

-----------------------------------------------------------------------
Next ,  execute the following on the SOURCE schema;
conn source_schema/password@source_db
variable p_Schema varchar2(30)
exec :p_Schema := sys_context('userenv','current_schema');

variable p_Dest_Schema varchar2(30)
exec :p_Dest_Schema := '&Destination';

variable p_Dest_Password varchar2(30)
exec :p_Dest_Password := '&Password';

variable p_Dest_SID varchar2(30)
exec :p_Dest_SID := '&SID';

exec dbms_aqadm.stop_queue(:p_Schema||'.AQ_EMS_QUEUE');
exec dbms_aqadm.drop_queue(:p_Schema||'.AQ_EMS_QUEUE');
begin
  dbms_aqadm.drop_queue_table(queue_table=>:p_Schema||'.T_AQ_EMS_QUEUE'
                             ,force=>true);
end;
/
/* Create a user defined data type first */
create or replace type ty_Msg as object
(id number,msg varchar2(100));
/
begin
  execute immediate
    'create database link d8_link connect to ' || :p_Dest_Schema
                                               || ' identified by '
                                               || :p_Dest_Password
                                               || ' using '''
                                               || :p_Dest_SID
                                               || '''';
end;
/

-- Run this to ensure the database link works.
-- If it doesn't then do not continue until the
-- database link can be sucessfully established
select  1 l_Num
from    dual@D8_LINK;

begin
  dbms_aqadm.create_queue_table(queue_table=>:p_Schema||'.T_AQ_EMS_QUEUE'
                               ,queue_payload_type=>'TY_MSG'
                               ,multiple_consumers=>true);
end;
/
begin
  dbms_aqadm.create_queue(queue_name=>:p_Schema||'.AQ_EMS_QUEUE'
                         ,queue_table =>:p_Schema||'.T_AQ_EMS_QUEUE');
end;
/
-- associate a subscriber with the queue
set serveroutput on
begin
  dbms_aqadm.add_subscriber
    (queue_name=>:p_Schema||'.AQ_EMS_QUEUE'
                ,subscriber=>sys.aq$_agent
                                          (
                                           'SUB1'
                                          ,:p_Dest_Schema
                                           ||'.AQ_EMS_QUEUE@D8_LINK'
                                          ,0)
                           ,queue_to_queue  => true);
  dbms_output.put_line('Subscriber sucessfully set');
end;
/
exec dbms_aqadm.start_queue(queue_name=>:p_Schema||'.AQ_EMS_QUEUE');

-- Run the following script to verify that the SOURCE and
-- DESTINATION queue's match
-- the query should be run on the SOURCE schema
set serveroutput on
declare
  l_RC        binary_integer := 0;
  l_db_Link   all_db_links.db_link%type;
  l_Continue  boolean := false;
begin

  -- make sure you get the full domain name
  begin
    select  db_link
    into    l_db_Link
    from    all_db_links
    where   owner     = :p_Schema
    and     db_link   like 'D8_LINK%';

    exception
      when too_many_rows then
        l_Continue := false;
      when no_data_found then
        l_Continue := false;
  end;

  l_Continue := (l_db_Link is not null);

  if (l_Continue) then
    -- The queue names for source and desitination are the same,
    -- the only difference being the schema names
    dbms_aqadm.verify_queue_types
      (src_queue_name   => :p_Schema||'.AQ_EMS_QUEUE'
      ,dest_queue_name  => :p_Dest_Schema||'.AQ_EMS_QUEUE'
      ,destination      => l_db_Link -- Database link name
      ,rc               => l_RC);

    dbms_output.put_line(case l_RC
                          when 0 then
                            'Queues do not match'
                          when 1 then
                            'Queues match'
                          else
                            'Error determining status'
                         end);
  else
    dbms_output.put_line('Unknown DB Link,Verification unsuccessful');
  end if;

end;
/

set serveroutput on
declare
  l_db_Link   all_db_links.db_link%type;
  l_Continue  boolean := false;
  PROPAGATION_EXISTS    exception;
  pragma                exception_init(PROPAGATION_EXISTS, -24041);

begin

  -- make sure you get the full domain name
  begin
    select  db_link
    into    l_db_Link
    from    all_db_links
    where   owner     = :p_Schema
    and     db_link   like 'D8_LINK%';

    exception
      when too_many_rows then
        l_Continue := false;
      when no_data_found then
        l_Continue := false;
  end;

  l_Continue := (l_db_Link is not null);

  if (l_Continue) then
    dbms_aqadm.schedule_propagation
      (queue_name        => :p_Schema||'.AQ_EMS_QUEUE'
      ,destination_queue => :p_Dest_Schema||'.AQ_EMS_QUEUE'
      ,destination       => l_db_Link
      ,start_time        => sysdate--SYSDATE indicate immediate
      ,duration          => null   --propagation until stopped
      ,latency           => 0);    --Indicates gap before propagating

    dbms_output.put_line('Propagation successfully scheduled ');
  else
    dbms_output.put_line('Unknown DB Link,Propagation not scheduled');
  end if;

  exception
    when PROPAGATION_EXISTS then
      dbms_output.put_line('Propagation schedule already exists');

end;
/

prompt Test enqueue ,but only after successfully starting scheduling propagation
set serveroutput on
declare
  enqueue_options     dbms_aq.enqueue_options_t;
  message_properties  dbms_aq.message_properties_t;
  message_handle      raw(16);
  message             ty_Msg := ty_Msg(1,'test 1 AQ demo 3');
begin
  dbms_aq.enqueue(queue_name        =>:p_Schema||'.AQ_EMS_QUEUE'
                 ,enqueue_options   =>enqueue_options
                 ,message_properties=>message_properties
                 ,payload           =>message
                 ,msgid             =>message_handle);
  commit;

  dbms_output.put_line('Message sucessfully enqueued ' );
  exception
    when others then
      dbms_output.put_line('Failure ' || sqlerrm);

end;
/

-- This needs to be run when there have been multiple failures with
-- propogation due to incorect configuration parameters.
-- Oracle automatically disables the propagation in such cases
-- for security reasons.
set serveroutput on
declare

  NO_PROPAGATION_EXISTS exception;
  pragma                exception_init(NO_PROPAGATION_EXISTS, -24042);

  l_db_Link   all_db_links.db_link%type;
  l_Continue  boolean := false;
begin

  -- make sure you get the full domain name
  begin
    select  db_link
    into    l_db_Link
    from    all_db_links
    where   owner     = :p_Schema
    and     db_link   like 'D8_LINK%';

    exception
      when too_many_rows then
        l_Continue := false;
      when no_data_found then
        l_Continue := false;
  end;

  l_Continue := (l_db_Link is not null);

  if (l_Continue) then

    dbms_aqadm.enable_propagation_schedule
      (queue_name         => :p_Schema||'.AQ_EMS_QUEUE'
      ,destination_queue  => :p_Dest_Schema||'.AQ_EMS_QUEUE'
      ,destination        => l_db_Link);

    dbms_output.put_line('Propagation successfully enabled ');
  else
    dbms_output.put_line('Unknown DB Link,Propagation not enabled');
  end if;

  exception
    when NO_PROPAGATION_EXISTS then
      dbms_output.put_line('No propagation exists between queues');
end;
/

-- To unschedule propagation
set serveroutput on
declare

  NO_PROPAGATION_EXISTS exception;
  pragma                exception_init(NO_PROPAGATION_EXISTS, -24042);

  l_db_Link   all_db_links.db_link%type;
  l_Continue  boolean := false;
begin

  -- make sure you get the full domain name
  begin
    select  db_link
    into    l_db_Link
    from    all_db_links
    where   owner     = :p_Schema
    and     db_link   like 'D8_LINK%';

    exception
      when too_many_rows then
        l_Continue := false;
      when no_data_found then
        l_Continue := false;
  end;

  l_Continue := (l_db_Link is not null);

  if (l_Continue) then

    dbms_aqadm.unschedule_propagation
      (queue_name         => :p_Schema||'.AQ_EMS_QUEUE'
      ,destination_queue  => :p_Dest_Schema||'.AQ_EMS_QUEUE'
      ,destination        => l_db_Link);

    dbms_output.put_line('Propagation successfully stopped');
  else
    dbms_output.put_line('Unknown DB Link,Propagation not unscheduled');
  end if;

  exception
    when NO_PROPAGATION_EXISTS then
      dbms_output.put_line('No propagation exists between queues');
end;
/

-- To stop propagation
set serveroutput on
declare

  NO_PROPAGATION_EXISTS exception;
  pragma                exception_init(NO_PROPAGATION_EXISTS, -24042);

  l_db_Link   all_db_links.db_link%type;
  l_Continue  boolean := false;
begin

  -- make sure you get the full domain name
  begin
    select  db_link
    into    l_db_Link
    from    all_db_links
    where   owner     = :p_Schema
    and     db_link   like 'D8_LINK%';

    exception
      when too_many_rows then
        l_Continue := false;
      when no_data_found then
        l_Continue := false;
  end;

  l_Continue := (l_db_Link is not null);

  if (l_Continue) then

    dbms_aqadm.disable_propagation_schedule
      (queue_name        => :p_Schema||'.AQ_EMS_QUEUE'
      ,destination_queue => :p_Dest_Schema||'.AQ_EMS_QUEUE'
      ,destination       => l_db_Link);

    dbms_output.put_line('Propagation successfully stopped');
  else
        dbms_output.put_line('Unknown DB Link,Propagation not disabled');
  end if;


  exception
    when NO_PROPAGATION_EXISTS then
      dbms_output.put_line('No propagation exists between queues');

end;
/

-- to purge a queue table, use the following;
set serveroutput on
declare
  po dbms_aqadm.aq$_purge_options_t;
begin

  po.block := true;

  dbms_aqadm.purge_queue_table
    (queue_table     =>:p_Schema||'.AQ_EMS_QUEUE'
    ,purge_condition =>null
    ,purge_options   =>po);

  dbms_output.put_line('Purge sucessful');
  exception
    when others then
      dbms_output.put_line('Fail ' || sqlerrm);
end;
/

/*

Troubleshooting propogation
---------------------------
1.  Ensure the job_queue_processes initialization parameter
    is greater than zero. If it is not greater than zero,
    then unschedule propagation, before setting it
    to a value greater than zero.

2.  Ensure database link is active, i.e.
       select 1 from dual@database_link_name
    should return 1

3.  Ensure the payload is the same on both instances

4.  Verify that the queues match using the
    DBMS_AQADN.VERIFY_QUEUE_TYPES built-in procedure

5.  Verify that propagation is currently not disabled,
    using the following SQL;

    select schedule_disabled from  user_queue_schedules;
    should return "N"

6.  Check the NEXT_RUN_DATE, NEXT_RUN_TIME and PROPAGATION_WINDOW
    on USER_QUEUE_SCHEDULES. If the date and time are in the past and
    remain in the past, then the propagation schedule has been
    incorrectly configured.

    If the propagation window is null then the DURATION parameter may
    have been set to NULL, which is correct, but ensure that this is
    in fact the case.

    use the following SQL;
      select next_run_date, next_run_time,PROPAGATION_WINDOW
      from user_queue_schedules;

7.  Ensure JOBNO is not null in SYS.AQ$_SCHEDULES using the
    following SQL;
      select jobno from sys.aq$_schedules;

8.  Verify that the AQ_TM_PROCESSES is zero

9.  Verify that the AQ


Short Script
col next_run_date format a40
col next_run_time format a20
col owner format a20
col object_table format a30
col object_type format a40
col type format a20
col name format a20
col value format a20
set linesize 400
set pages 2000


select jobno
from   sys.aq$_schedules
/
select schedule_disabled
from   user_queue_schedules
/
select next_run_date
      ,next_run_time
      ,propagation_window
from user_queue_schedules
/
select count(*) from sys.job$
/
select name
       ,value
from   v$parameter
where name in('job_queue_processes','aq_tm_processes')
/
select owner
      ,queue_table
      ,type
      ,object_type
from dba_queue_tables
/


-- To trace for AQ problems
alter session set events '24040 trace name context forever, level 10';

*/

Oracle JMS Dequeue

Read and test the enqueue before attempting the dequeue;
http://bluefrog-oracle.blogspot.com/2011/10/oracle-jms-enqueue.html 
SQL> create or replace and compile java source named BasicAQDequeue as
  2  import java.sql.Connection;
  3  import java.sql.Driver;
  4  import java.sql.DriverManager;
  5  import javax.jms.Queue;
  6  import javax.jms.QueueConnection;
  7  import javax.jms.QueueConnectionFactory;
  8  import javax.jms.QueueReceiver;
  9  import javax.jms.QueueSession;
 10  import javax.jms.Message;
 11  import javax.jms.TextMessage;
 12  import javax.jms.Session;
 13  import oracle.jms.AQjmsFactory;
 14  import java.sql.PreparedStatement;
 15  import java.sql.ResultSet;
 16  import oracle.jdbc.*;
 17  import oracle.jdbc.aq.*;
 18  import oracle.jms.*;
 19 
 20  public class BasicAQDequeue {
 21 
 22      private QueueReceiver     consumer;
 23      private Queue             queue;
 24      private QueueConnection   consumerConnection;
 25      private QueueSession      consumerSession;
 26      private Connection        conn;
 27      private static final String   QUEUE_NAME  = "JMS_TEXT_QUE";
 28      private static final int      ACK_MODE    = QueueSession.AUTO_ACKNOWLEDGE;
 29      private static final boolean  TRANSACTED  = true;
 30 
 31      /* Creates and initializes the test. */
 32      public BasicAQDequeue() throws Exception {
 33          super();
 34          // init consumer session - Use Oracle default thin driver connection
 35          conn                = DriverManager.getConnection("jdbc:default:connection:");
 36          consumerConnection  = AQjmsQueueConnectionFactory.createQueueConnection(conn);
 37          consumerConnection.start();
 38          consumerSession     = consumerConnection.createQueueSession(TRANSACTED,ACK_MODE);
 39          queue               = consumerSession.createQueue(QUEUE_NAME);
 40          consumer            = consumerSession.createReceiver(queue);
 41      }
 42      public static void main (String args[]) throws Exception
 43        {
 44          BasicAQDequeue qTest  = new BasicAQDequeue();
 45          qTest.run();
 46          qTest.cleanup();
 47        }
 48      /** Closes the JMS connections.throws Exception */
 49      private void cleanup() throws Exception {
 50          // cleanup consumer
 51          System.out.println("Start cleanup.");
 52          consumerConnection.stop();
 53          consumer.close();
 54          consumerSession.close();
 55          consumerConnection.close();
 56          System.out.println("Done");
 57      }
 58      /* The message send. Throws Exception */
 59      private void run() throws Exception {
 60 
 61          TextMessage message;
 62          message = (TextMessage)consumer.receive(1000);
 63          consumerSession.commit();
 64          System.out.println("Message received: " + message.getText());
 65      }
 66  }
 67  /

Java created.

SQL> create or replace procedure BasicAQDequeue as
  2  language java name 'BasicAQDequeue.main(java.lang.String[])';
  3  /

Procedure created.

SQL> exec dbms_java.set_output(10000);

PL/SQL procedure successfully completed.

SQL> set serveroutput on
SQL> begin
  2    BasicAQEnqueue;
  3  end;
  4  /
Message sent: Test Message 1
Start cleanup.
Done

PL/SQL procedure successfully completed.

Oracle JMS Enqueue

The following performs a test enqueue to an Oracle AQ queue using a JMS session executed within the Oracle database;

First, ensure the schema that the script is executed on has the following privileges granted from SYS;

connect sys/password@orcl as sysdba
grant AQ_ADMINISTRATOR_ROLE to jms_schema
/
grant aq_user_role to jms_schema
/
grant JAVA_ADMIN, JAVAUSERPRIV, JAVASYSPRIV, JAVADEBUGPRIV to jms_schema
/
grant execute on DBMS_AQIN to jms_schema;
/
grant execute on DBMS_JMS_PLSQL to jms_schema
/
Next, ensure a queue with a JMS_TEXT_MESSAGE payload exists, use the following

-- login with the schema you wish to run the demo on, for example JMS_SCHEMA

exec dbms_aqadm.create_queue_table (queue_table=>'jms_qtt_text', queue_payload_type=>'SYS.AQ$_JMS_TEXT_MESSAGE');
exec dbms_aqadm.create_queue (Queue_name=>'jms_text_que',Queue_table=>'jms_qtt_text');
exec dbms_aqadm.create_queue_table (queue_table=>'jms_qtt_text', queue_payload_type=>'SYS.AQ$_JMS_TEXT_MESSAGE');

Ready to create and compile the Java source;

SQL> create or replace and compile java source named BasicAQEnqueue as
  2  import java.sql.Connection;
  3  import java.sql.Driver;
  4  import java.sql.DriverManager;
  5  import javax.jms.Queue;
  6  import javax.jms.QueueConnection;
  7  import javax.jms.QueueSender;
  8  import javax.jms.QueueSession;
  9  import oracle.jms.AQjmsFactory;
 10  import oracle.jdbc.*;
 11  import oracle.jdbc.aq.*;
 12  import oracle.jms.*;
 13 
 14  public class BasicAQEnqueue {
 15 
 16      private QueueSender       producer;
 17      private Queue             queue;
 18      private QueueConnection   producerConnection;
 19      private QueueSession      producerSession;
 20      private static final String   QUEUE_NAME  = "JMS_TEXT_QUE";
 21      private static final int      ACK_MODE    = QueueSession.AUTO_ACKNOWLEDGE;
 22      private static final boolean  TRANSACTED  = true;
 23 
 24      /* Creates and initializes the test. */
 25      public BasicAQEnqueue() throws Exception {
 26          super();
 27          // init producer session - Use Oracle default thin driver connection
 28          Connection conn     = DriverManager.getConnection("jdbc:default:connection:");
 29          producerConnection  = AQjmsQueueConnectionFactory.createQueueConnection(conn);
 30          producerSession     = producerConnection.createQueueSession(TRANSACTED,ACK_MODE);
 31          queue               = producerSession.createQueue(QUEUE_NAME);
 32          producer            = producerSession.createSender(queue);
 33      }
 34 
 35      public static void main (String args[]) throws Exception
 36        {
 37          BasicAQEnqueue qTest  = new BasicAQEnqueue();
 38          qTest.run();
 39          qTest.cleanup();
 40        }
 41 
 42      /** Closes the JMS connections.throws Exception */
 43      private void cleanup() throws Exception {
 44          // cleanup producer
 45          System.out.println("Start cleanup.");
 46          producerConnection.stop();
 47          producer.close();
 48          producerSession.close();
 49          producerConnection.close();
 50          System.out.println("Done");
 51      }
 52 
 53      /* The message send. Throws Exception */
 54      private void run() throws Exception {
 55          String msg = "Test Message 1";
 56          producer.send(producerSession.createTextMessage(msg));
 57          producerSession.commit();
 58          System.out.println("Message sent: " + msg);
 59      }
 60  }
 61  /

Java created.


SQL> create or replace procedure BasicAQEnqueue as
  2  language java name 'BasicAQEnqueue.main(java.lang.String[])';
  3  /

Procedure created.

SQL> set serveroutput on
SQL> begin
  2    BasicAQEnqueue;
  3  end;
  4  /
Message sent: Test Message 1
Start cleanup.
Done

SQL> -- To view the messages in SQL Plus, use the following SQL;
SQL> col Msg format a20
SQL> select msg_id, msg_state, qt.user_data.text_vc Msg
  2  from aq$jms_qtt_text qt
  3  /

MSG_ID                           MSG_STATE        MSG
-------------------------------- ---------------- --------------
E39998B62A5046DCA6F2316D64EF3048 READY            Test Message 1

JMS connection within an Oracle instance

The schema must have the following privileges;
granted from SYS/password@orcl as sysdba to schema

connect sys/password@orcl as sysdba

grant AQ_ADMINISTRATOR_ROLE to jms_schema
/
grant aq_user_role to jms_schema
/
grant JAVA_ADMIN

     ,JAVAUSERPRIV

     ,JAVASYSPRIV

     ,JAVADEBUGPRIV 

to jms_schema
/
grant execute on DBMS_AQIN to jms_schema;
/
grant execute on DBMS_JMS_PLSQL to jms_schema
/

connect as jms_schema/password@orcl

SQL> create or replace and compile java source named BasicAQConnection as
  2  import java.sql.Connection;
  3  import java.sql.Driver;
  4  import java.sql.DriverManager;
  5  import java.sql.PreparedStatement;
  6  import java.sql.ResultSet;
  7  import javax.jms.QueueConnection;
  8  import javax.jms.QueueConnectionFactory;
  9  import javax.jms.QueueSession;
 10  import javax.jms.Session;
 11  import oracle.jms.AQjmsFactory;
 12  import oracle.jdbc.*;
 13  import oracle.jdbc.aq.*;
 14  import oracle.jms.*;
 15 
 16  public class BasicAQConnection
 17    {
 18      public static void main (String args[]) throws Exception
 19        {
 20          Connection conn = DriverManager.getConnection("jdbc:default:connection:");
 21          QueueConnection qc = AQjmsQueueConnectionFactory.createQueueConnection(conn);
 22          QueueSession qs = qc.createQueueSession(true, Session.CLIENT_ACKNOWLEDGE);
 23 
 24          PreparedStatement pStmt = conn.prepareStatement("select count(*) from emp");
 25          ResultSet rSet = pStmt.executeQuery();
 26          rSet.next();
 27          System.out.println("Number of Employees is: " + rSet.getString(1));
 28          rSet.close();
 29 
 30          qc.stop();
 31          conn.close();
 32          qc.close();
 33 
 34        }
 35    }
 36  /


Java created.


SQL>
SQL> create or replace procedure BasicAQConnection as
  2  language java name 'BasicAQConnection.main(java.lang.String[])';
  3  /


Procedure created.


SQL>
SQL> exec dbms_java.set_output(10000);


PL/SQL procedure successfully completed.


SQL>
SQL> select count(*) from emp
  2  /


  COUNT(*)
----------
         1


SQL> set serveroutput on
SQL> begin
  2    BasicAQConnection;
  3  end;
  4  /
Number of Employees is: 1


PL/SQL procedure successfully completed.

View constraints in a hierarchy

To view a hierarchy of all constraints, starting with a Master table, use the following SQL statement;

select lpad(' ', 4 * level - 1 ) || sys_connect_by_path(Child, '/') as Hierarchy
from
  (
    select  ac.owner || '.' ||  ac.table_name as Child
           ,r.owner  || '.' ||  r.table_name  as Master
    from    all_constraints     ac
           ,all_constraints     r
    where   r.Owner             = ac.r_Owner
    and     r.Constraint_Name   = ac.r_Constraint_Name
    and     ac.Constraint_Type  = 'R'
)
start with Master =  sys_context('userenv','current_schema')||'.CLIENT'
connect by prior Child =  Master
/

For example, given the following set of tables and foreign keys;

create table product (p_id          number
                     ,product_name  varchar2(10)
                     ,price number);
alter table product add constraint p_pk primary key (p_id);

create table client  (c_id number
                     ,client_name varchar2(10));
alter table client add constraint c_pk primary key (c_id);

create table client_order (co_id  number
                          ,c_id   number
                          ,d      date);
alter table client_order add constraint co_pk primary key (co_id);

alter table client_order add constraint co_fk
foreign key (c_id) references client (c_id);

create table order_line (ol_id      number
                        ,co_id      number
                        ,p_id       number
                        ,qty        number
                        ,ol_amount  number);
alter table order_line add constraint ol_pk primary key (ol_id);

alter table order_line add constraint ol_cl_fk
foreign key (co_id) references client_order (co_id);

alter table order_line add constraint ol_po_fk
foreign key (p_id) references product (p_id);

Running the SQL against the CLIENT table will produce the following result;

SQL> variable cs varchar2(30)

SQL> exec :cs := sys_context('userenv','current_schema');

PL/SQL procedure successfully completed.

SQL> select    lpad(' ', 4 * level - 1 )

  2         || sys_connect_by_path(Child, '/') as Hierarchy

  3  from

  4    (

  5      select  ac.owner || '.' ||  ac.table_name as Child

  6             ,r.owner  || '.' ||  r.table_name  as Master

  7      from    all_constraints     ac

  8             ,all_constraints     r

  9      where   r.Owner             = ac.r_Owner

 10      and     r.Constraint_Name   = ac.r_Constraint_Name

 11      and     ac.Constraint_Type  = 'R'

 12  )

 13  start with Master=:cs||'.CLIENT'

 14  connect by prior Child =  Master;

HIERARCHY

------------------------------------------------------------

   /XDB_DEV4.CLIENT_ORDER

       /XDB_DEV4.CLIENT_ORDER/XDB_DEV4.ORDER_LINE

The following query lists the path leading to each table and the level at which the set of foreign key s exists within the path.

with t as
(
    select  Ref_Level
           ,Path
           ,substr(Path
                   ,instr(Path, '/', 1,Ref_Level) +1
                   , length(Path) ) as Table_Name
    from
    (
    select distinct
           level as ref_level
          ,lpad(' ', 4 * level - 1 ) 
               || sys_connect_by_path(Master, '/') as Path
    from
      (
        select  ac.table_name       as Child
               ,r.table_name        as Master
        from   
                all_constraints     ac
               ,all_constraints     r
        where  
                r.Owner             = ac.r_Owner
        and     r.Owner             = '&SCHEMA'
        and     r.Constraint_Name   = ac.r_Constraint_Name
        and     ac.Constraint_Type  = 'R'
    )
    connect by nocycle prior Child = Master
    )
)
--
select *
from
(
select
        Ref_Level
       ,nvl(first_value(Ref_Level) 
        over (partition by table_name 
              order by rn), Ref_Level) as last_level
       ,Path
       ,Table_Name
       ,rn
from
    (
    select     
            Ref_Level
           ,Path 
           ,Table_Name
           ,row_number() 
            over (partition by table_name 
                  order by Ref_Level desc) as rn
    from
            t
    )
);

Listing all foreign key constraints

select ac.owner
      ,ac.table_name
      ,r.owner        as ref_owner
      ,r.table_name   as ref_table
from   all_constraints ac
      ,all_constraints r 
where  r.owner            = ac.r_owner
and    r.constraint_name  = ac.r_constraint_name
and    ac.constraint_type = 'R'
/

pipelined Date function

The requirement for this function is to
list the months ahead of the month supplied as parameters (both  numbers).
The number of months ahead of the month supplied would also
be supplied as a parameter.


create or replace type dtTable as table of date;
/
create or replace function p (pYear       in number
                             ,pMonth      in number
                             ,pMth_Ahead  in pls_integer) 

  return dtTable pipelined as

 l_In_Date date := to_date(to_char(pYear)||to_char(pMonth), 'yyyymm');

begin

  for i in (select add_months(l_In_Date,(level-1)) dt
            from dual
            connect by level <= pMth_Ahead) loop
    pipe row(i.dt);
  end loop;
  return;
end;
/
SQL> select p.column_Value as Month from  table(p(2008,9,6)) p;

MONTH
---------
01-SEP-08
01-OCT-08
01-NOV-08
01-DEC-08
01-JAN-09
01-FEB-09

6 rows selected.

OUTER Join combined with INSERT ALL

The following is an example of an INSERT ALL. The requirement is that rows that exists in table A be inserted into table B, but only if they exist in the MASTER table. If not, then insert an error message into the LOG_MSG table.

create table a (
col1 number
,col2 varchar2(3)
,col3 varchar2(3)
)
/

create table b (
col1 number
,col2 varchar2(3)
,col3 varchar2(3)
)
/

create table master (
col1 varchar2(3)
,col2 number
)
/

create table log_msg
(msg varchar2(100)
,t timestamp default current_timestamp)
/


insert into a (col1,col2,col3)
  select 1 col1, 'AAA' col2,  'P01' col3 from dual union all
  select 2, 'BBB', 'Q01' from dual union all
  select 3, 'CCC', 'S05' from dual union all
  select 4, 'DDD', 'A02' from dual union all
  select 5, 'EEE', 'B01' from dual union all
  select 6, 'FFF', 'B03' from dual
/

insert into b (col1,col2,col3)
select 1 col1, 'AAA' col2,  'P01' col3 from dual union all
  select 2, 'BBB', 'Q01' from dual union all
  select 3, 'CCC', 'S05' from dual union all
  select 5, 'EEE', 'B01' from dual
/


insert into master (col1,col2)
select 'P01', 100 from dual union all
select 'P02', 400 from dual union all
select 'Q01', 900 from dual union all
select 'Q02', 100 from dual union all
select 'S01', 800 from dual union all
select 'S05', 500 from dual union all
select 'B01', 200 from dual union all
select 'B03', 900 from dual
/

insert all
  when Master_ID is not null then
    into b (col1, col2, col3) values (col1, col2, col3)
  when Master_ID is null then
    into log_msg (msg) values (col1 || ' ' || col2 || ' ' || col3 || ' does not exist on Master')
select m.col1 as Master_ID
      ,a.col1 as Col1
      ,a.col2 as Col2
      ,a.col3 as Col3
from (
      select a.col1, a.col2, a.col3
      from a
      left outer join b on  a.col2 = b.col2
                        and a.col3 = b.col3
      where b.col1 is null
    ) a
left outer join master m on a.col3 = m.col1
/

SQL> select * from b;

      COL1 COL COL
---------- --- ---
         1 AAA P01
         2 BBB Q01
         3 CCC S05
         5 EEE B01
         6 FFF FFF

SQL> select * from log_msg;

MSG                             T
------------------------------- -------------------------
4 DDD A02 does not exist Master 28-SEP-11 11.22.09.221000