Thanks. If the replicated table can have the _commit_version in strict sequence, I can take it as a global ever-incremental col and consume the delta of it (e.g. in batch way) with 

select * from replicated_tgt where _commit_version > (
    selecct last_version_offset = max(_commit_version) from downstream
)

Thanks.