How Does the Streams Capture Follow Up The Applied Changes?

Didn't you ever wonder how an asynchronous Streams capture gets its feedback from the apply processes? For weeks it did not make any sense to me! That was obvious the capture got some kind of feedback from the apply, though. The fact the only mandatory database links was going from the capture to the apply had turned that question into an even bigger mystery to me... This post shows there is actually no magic or hidden features involve. That's pretty simple.

I'll let you draw the full map of what happens when you have messages propagated into several databases and involving multiple applies. I'll focus on a very simple case that takes not more than a few minutes to build and is available in one of my previous post entitled: "Oracle Streams One Way Table Replication 101". This post is also a good pretext to introduce some of the streams processes as well as several ways to trace the associated operations...

Note:
What follows, has been tested with a 11.1.0.7 on Linux x86 32-bit database and that's very likely some of the described behaviors depend on the Oracle Release.

This post is made of the following sections:

• Why do I say the capture knows about the applied changes?
• Where is the applied SCN stored?
• What process changes these data?
• How does the capture process get data from the apply?
• Even more interesting findings/questions

Why do I say the capture knows about the applied changes?

First, it has to know! That's because, in the case of asynchronous capture, the messages are enqueued the buffered part of the queue which is not persistent. For the capture process to free the archived logs and provide a reliable mecanism in the case of a failure, it has to monitor the changes that have been applied on the destination. This way, if one instance crashes and logical change records in buffered queues are lost, the capture can restart from the point it knows the messages have been applied for sure.

Second, you can find some data regarding the apply on the capture part of your streams configuration. The APPLIED_SCN column of the DBA_CAPTURE view is incremented on a regular manner when changes are applied. This simple test, once you have the streams configuration running shows those changes:

select applied_scn
 from dba_capture;

APPLIED_SCN
-----------
   1010398

insert into source.t1
  values (4,'Text 4');

col id format 99
col text format a6

select *
  from destination.t1;

ID TEXT
-- ------
3 Text 3
4 Text 4
1 Text 1
2 Text 2

select applied_scn
 from dba_capture;

APPLIED_SCN
-----------
   1010398

So what we discover from that first test is that the last applied SCN is not propagated right away on the capture side of the database; actually if there is no change it can take a while. However if you wait and run the query again, you'll see changes are known from the capture side:

select applied_scn
 from dba_capture;

APPLIED_SCN
-----------
   1026804

Where is the applied SCN stored?

The next legitimate question is to figure out what is the underlying structure that stores the applied scn. To answer that question, we'll just dig into the Streams dictionary:

col text format a80
set long 10000
set longchunksize 10000

select text
 from dba_views
where owner='SYS'
  and view_name='DBA_CAPTURE';

TEXT
----------------------------------------------------------------------------------------------------
select cp.capture_name, cp.queue_name, cp.queue_owner, cp.ruleset_name,
      cp.ruleset_owner, u.name, cp.start_scn,
      decode(cp.status, 1, 'DISABLED',
    2, 'ENABLED',
    4, 'ABORTED', 'UNKNOWN'),
      cp.spare1, cp.spare2,
      decode(cp.use_dblink, 1, 'YES', 'NO'),
      cp.first_scn, cp.source_dbname, dl.source_dbid, dl.source_resetlogs_scn,
      dl.source_resetlogs_time, cp.logmnr_sid, cp.negative_ruleset_name,
      cp.negative_ruleset_owner,
-- MAX_CHECKPOINT_SCN, -1 for ckpt free
      decode(bitand(cp.flags, 8192), 8192, -1, nvl(dl.checkpoint_scn, 0)),
-- use applied_scn(spare2) for ckpt free
      decode(bitand(cp.flags, 8192), 8192, nvl(cp.spare2,0),
       dbms_logrep_util.get_req_ckpt_scn(dl.id, nvl(cp.spare2,0))),
      decode(bitand(cp.flags, 4), 4, 'IMPLICIT', 'EXPLICIT'),
      cp.status_change_time, cp.error_number,
      cp.error_message, cp.version,
      decode(bitand(cp.flags, 64), 64, 'DOWNSTREAM', 'LOCAL'),
      dbms_logrep_util.get_last_enq_scn(cp.capture_name), cp.spare3
 from "_DBA_CAPTURE" cp,
      dba_logmnr_session dl,
      sys.user$ u
where dl.id (+) = cp.logmnr_sid
  and cp.capture_userid = u.user# (+)
  and (bitand(cp.flags,512) != 512) -- skip sync capture

From that view definition you can see the data come from _DBA_CAPTURE.SPARE2 which itself comes from streams$_capture_process.SPARE2 :

col text format a80
set long 10000
set longchunksize 10000

select text
 from dba_views
where owner='SYS'
  and view_name='_DBA_CAPTURE';

TEXT
----------------------------------------------------------------------------------------------------
select
 queue_oid, queue_owner, queue_name, capture#, capture_name,
 status, ruleset_owner, ruleset_name, logmnr_sid, predumpscn,
 dumpseqbeg, dumpseqend, postdumpscn, flags, start_scn, capture_userid,
 spare1, spare2, spare3, use_dblink, first_scn, source_dbname,
 spare4, spare5, spare6, spare7, negative_ruleset_owner,
 negative_ruleset_name, start_date, end_date, status_change_time,
 error_number, error_message, version
from sys.streams$_capture_process

What process changes these data?

We'll use the fact we've setup supplemental logging for the whole database to track the process that changes streams$_capture_process. In order to proceed, we'll perform the same test we did in the first part of this post but we'll archive the redo log before and after the test and we'll use LogMiner to get the information.

delete from source.t1;

commit;

alter system archive log current;

select thread#, max(sequence#)
 from v$archived_log
group by thread#;

THREAD# MAX(SEQUENCE#)
------- --------------
     1             17

select applied_scn
 from dba_capture;

APPLIED_SCN
-----------
   1081340

begin
 for i in 1..100000 loop
   insert into source.t1
      values (i,'Text '||to_char(i));
   commit;
 end loop;
end;
/

select count(*)
  from destination.t1;

COUNT(*)
--------
  47895

/
COUNT(*)
--------
  81010

/
COUNT(*)
--------
 100000

select applied_scn
 from dba_capture;

APPLIED_SCN
-----------
   1081340

Once the change made, we'll wait for a few minutes so that the applied SCN is changed on the capture side. Once we are sure that's the case, we'll archive the redo log again and get the file names that contain the transactions:

select applied_scn
 from dba_capture;

APPLIED_SCN
-----------
   1597521

alter system archive log current;

select thread#, max(sequence#)
 from v$archived_log
group by thread#;

THREAD# MAX(SEQUENCE#)
------- --------------
     1             25

select name from v$archived_log
 where thread#=1
   and sequence# between 18 and 25; 

NAME
------------------------------------------------------------
/u01/app/oracle/oradata/BLACK/archivelogs/1_18_679061156.dbf
/u01/app/oracle/oradata/BLACK/archivelogs/1_19_679061156.dbf
/u01/app/oracle/oradata/BLACK/archivelogs/1_20_679061156.dbf
/u01/app/oracle/oradata/BLACK/archivelogs/1_21_679061156.dbf
/u01/app/oracle/oradata/BLACK/archivelogs/1_22_679061156.dbf
/u01/app/oracle/oradata/BLACK/archivelogs/1_23_679061156.dbf
/u01/app/oracle/oradata/BLACK/archivelogs/1_24_679061156.dbf
/u01/app/oracle/oradata/BLACK/archivelogs/1_25_679061156.dbf

Once we know the data have been changed, we can query the archive logs for the change on streams$_capture_process:

begin
  dbms_logmnr.add_logfile(
     logfilename => '/u01/app/oracle/oradata/BLACK/archivelogs/1_18_679061156.dbf',
     options => dbms_logmnr.new);
  for i in 19..25 loop
     dbms_logmnr.add_logfile(
         logfilename => '/u01/app/oracle/oradata/BLACK/archivelogs/1_'||to_char(i)||'_679061156.dbf',
         options => dbms_logmnr.addfile);
  end loop;
end;
/

exec dbms_logmnr.start_logmnr( options => -
     dbms_logmnr.dict_from_online_catalog);

select operation, session#, serial#, count(*)
 from  v$logmnr_contents
where seg_owner='SYS' and seg_name='STREAMS$_CAPTURE_PROCESS'
group by operation, session#, serial#;

OPERATION SESSION# SERIAL# COUNT(*)
--------- -------- ------- --------
UPDATE         131       1        2

The process that does the change is the capture process as you can see it from below; to capture more changes, we'll trace that process with dbms_monitor:

select sid, serial#   
 from v$streams_capture;

SID SERIAL#
--- -------
131       1

exec dbms_monitor.session_trace_enable(131,1,true,true,'all_executions')

select tracefile
 from v$session s, v$process p
where s.sid=131    
  and p.addr=s.paddr;

TRACEFILE
----------------------------------------------------------------
/u01/app/oracle/diag/rdbms/black/BLACK/trace/BLACK_cp01_8702.trc

How does the capture process get data from the apply?

Once we've setup the traces, we can just wait for the update on streams$_capture_process to be executed and check the content of the trace file. You'll discover from those traces that the capture process queries streams$_apply_milestone that reside on the apply side of the configuration. If we restart a logminer session on the apply side this time and look for the session that update the streams$_apply_milestone table, you'll find that, it's the apply processes doing it. That table is actually updated by the apply reader and server processes:

exec dbms_monitor.session_trace_disable(131,1)

[...]

select operation, session#, serial#, count(*)
 from  v$logmnr_contents
where seg_owner='SYS' and seg_name='STREAMS$_APPLY_MILESTONE'
group by operation, session#, serial#;

OPERATION SESSION# SERIAL# COUNT(*)
--------- -------- ------- --------
UPDATE         128       1        4
UPDATE         126       1      109

select sid, serial# from v$streams_apply_server;

      SID SERIAL#
---------- -------
      126  1

select sid, serial# from v$streams_apply_reader;

      SID SERIAL#
---------- -------
      128  1

Even more interesting findings/questions...

Now that we've discovered how the capture keeps track of the apply changes by querying over the database link, we can use the same strategy to investigate other Streams internals in more advanced configurations. That leads to very interesting findings to; I'll show you one right away! If you check for the process applying the changes, you'll find the Apply Server Process; Restart a logMiner session on the apply side:

[...]

select operation, session#, serial#, count(*)
 from  v$logmnr_contents
where seg_owner='DESTINATION' and seg_name='T1'
group by operation, session#, serial#;

OPERATION SESSION# SERIAL# COUNT(*)
--------- -------- ------- --------
INSERT        126        1   100000

And if you trace it, you'll be able to catch the change made to streams$_apply_milestone, but nothing regarding the destination database:

exec dbms_monitor.session_trace_enable(126,1,true,true,'all_executions')

delete from source.t1;

commit;

select * from destination.t1;

no rows selected

select tracefile
 from v$session s, v$process p
where s.sid=126    
  and p.addr=s.paddr;

TRACEFILE
----------------------------------------------------------------
/u01/app/oracle/diag/rdbms/black/BLACK/trace/BLACK_as02_8708.trc

Actually if you want to track the change made to the table by the process you'll have to set the trace_level parameter like below:

exec dbms_apply_adm.set_parameter('streams_apply', 'trace_level',255);

begin
 for i in 1..100000 loop
   insert into source.t1
      values (i,'Text '||to_char(i));
   commit;
 end loop;
end;
/

If you edit the file you'll find the change records (LCR) that that are applied to the destination table but nothing like the SQL statements... Neither in V$SQL. or V$active_session_history. I guess that post raise more questions than it answers!