User-Defined DDL LCRs

In "User-Defined DML LCRs", I've shown how to use lcr$_row_record.construct to build LCRs for DELETE, UPDATE and INSERT; In the same post, I've created an apply process to execute those LCRs. I wanted to add an example of a DDL LCR at the same time, but because of an error with my 11.1.0.7 AL32UTF8 database, I have not been able to.

Not a big deal! I've created a WE8ISO8859P15 database and I'll correct that miss fire right away. You'll find below a sample configuration that shows how to use lcr$_ddl_record.construct.

Streams Queue and Apply

We'll use the same schema for the queue and for the apply process and to apply the DDL. To speed up the setup, I've granted DBA to the schema owner; There are 3 points worth to mention:

• I've set apply_capture parameter to false to make sure the apply process dequeues persistent messages
• I did not add any rule set to the apply so that it dequeues every message in the queue
• There is no need to define an instantiation SCN or to setup a SCN in the LCR (In the case of user-defined LCRs)

begin
 dbms_streams_adm.set_up_queue(
      queue_table => 'custom_queue_table',
      queue_name  => 'custom_queue');
end;
/
declare
 v_name varchar2(256);
begin
  select value into v_name
    from v$parameter
   where name='db_unique_name';
dbms_apply_adm.create_apply(
    queue_name => 'custom_queue',
    apply_name => 'custom_apply',
    apply_captured => false,
    source_database => v_name );
end;
/

exec dbms_apply_adm.start_apply('CUSTOM_APPLY');

Create and Enqueue A DDL LCR

Once everything ready you can build the LCRs and enqueue them in the queue so that the CREATE TABLE can be consumed and executed by the apply processes:

declare
  v_name varchar2(256);
  v_any     anydata;
  lcr       sys.lcr$_ddl_record;
  rc        pls_integer;
  enqopt    DBMS_AQ.ENQUEUE_OPTIONS_T;
  mprop     DBMS_AQ.MESSAGE_PROPERTIES_T;
  enq_msgid RAW(16);
begin
  -- Get DB Name and SCN
  select value into v_name
    from v$parameter
   where name='db_unique_name';

 mprop.SENDER_ID := SYS.AQ$_AGENT(user, null, null);

 lcr:=sys.lcr$_ddl_record.construct(
            source_database_name => v_name,
            command_type         => 'CREATE TABLE',
            object_owner         => user,
            object_name          => 'MYTABLE',
            object_type          => 'TABLE',
            ddl_text             =>
              'create table MYTABLE(id number primary key)',
            logon_user           => user,
            current_schema       => user,
            base_table_owner     => user,
            base_table_name      => 'MYTABLE',
            tag                  => null,
            transaction_id       => null,
            scn                  => null);

 DBMS_AQ.ENQUEUE(
   queue_name          =>  'custom_queue',
   enqueue_options     =>  enqopt,
   message_properties  =>  mprop,
   payload             =>  anydata.ConvertObject(lcr),
   msgid               =>  enq_msgid);
end;
/

commit;

desc mytable

Name Null?    Type
---- -------- ------
ID   NOT NULL NUMBER

declare
  v_name varchar2(256);
  v_any     anydata;
  lcr       sys.lcr$_ddl_record;
  rc        pls_integer;
  enqopt    DBMS_AQ.ENQUEUE_OPTIONS_T;
  mprop     DBMS_AQ.MESSAGE_PROPERTIES_T;
  enq_msgid RAW(16);
begin
  -- Get DB Name and SCN
  select value into v_name
    from v$parameter
   where name='db_unique_name';

 mprop.SENDER_ID := SYS.AQ$_AGENT(user, null, null);

 lcr:=sys.lcr$_ddl_record.construct(
            source_database_name => v_name,
            command_type         => 'DROP TABLE',
            object_owner         => user,
            object_name          => 'MYTABLE',
            object_type          => 'TABLE',
            ddl_text             =>
               'drop table MYTABLE purge',
            logon_user           => user,
            current_schema       => user,
            base_table_owner     => user,
            base_table_name      => 'MYTABLE',
            tag                  => null,
            transaction_id       => null,
            scn                  => null);
 DBMS_AQ.ENQUEUE(
   queue_name          =>  'custom_queue',
   enqueue_options     =>  enqopt,
   message_properties  =>  mprop,
   payload             =>  anydata.ConvertObject(lcr),
   msgid               =>  enq_msgid);
end;
/

commit;

desc mytable

ERROR:
ORA-04043: object mytable does not exist

Conclusion

This is it. With a WE8ISO8859P15 database, it works both with 10g and 11g and it should even work with Oracle Standard Edition... At least in theory!