Support add/drop tables

[zhousun]

Right now, pg_replicate will calling BatchDataPipeline::start, it will keep consuming cdc stream in a loop, and there's no way to add a table or drop a table from the stream.

1. We need a way for pg_mooncake to add a table to subscriber.

• Currently we are fetching table metadata on BatchDataPipeline::start by querying postgres, probably need a new way.

2. For replicating existing heap table, we need a way to 'backfill' before start applying cdc.