support crossdb query

[st1page]

• support ad-hoc query

      SELECT * FROM db1.schema.t join db2.schema.t; 
  

  after we will support the frontend part, which mainly is in binder and
• support steaming

      CREATE materialized view mv as SELECT * FROM schema.t join db.schema.t; 
  

  The db.schema.t's changes need to be read from the log store and achieve isolation

[wenym1]

support ad-hoc query
SELECT * FROM db1.schema.t join db2.schema.t;
after we will support the frontend part, which mainly is in binder and

What will be the expected epoch to be used in cross database ad-hoc query🤔

Since we won't ensure consistency between databases, I guess the intuitive one is that each database uses its own latest epoch. But currently we assume that all tables in a batch query shared a same batch query epoch. So this may require some refactors in the frontend batch query execution, to support different epoch for different tables.

[wenym1]

support steaming
CREATE materialized view mv as SELECT * FROM schema.t join db.schema.t;
The db.schema.t's changes need to be read from the log store and achieve isolation

Short note about the implementation of crossdb query.

Since we want to ensure failure isolation between databases, there won't be streaming dispatcher in the cross database stream table scan. In the initial version of cross database stream table scan, to read upstream data, we will only read the L0 log store of the upstream table.

The behavior is somehow like treating the upstream table as a source. Such source has different vnode as splits. The split offset is our consume progress in different vnodes. During backfill, we will consume the snapshot of the upstream table in a fixed epoch. In this stage, the consume offset under each vnode will be the pk of upstream table. After backfill, we will read the L0 log store of upstream table. In this stage, the consume offset under each vnode will be epoch + pk of upstream table. In general, we will use a separate state table to track the store the progress. The state table is similar to the one we use in recoverable backfill, with an extra epoch column, The schema of the state table will be

| vnode | epoch | is_finished | .. upstream pk|

Different to the current snapshot backfill, epoch in upstream table is only treated as an offset of consuming upstream table. We will receive barrier with independent epochs from the barrier manager. We may get the next upstream epoch to be consumed by an extra method call next_epoch in the StateStoreRead trait.