Page History

...

@PublicEvolving
public interface Catalog {
  
   /** Prepare replace table. */
   TwoPhaseCommitCatalogTableOptional<TwoPhaseCommitCatalogTable> prepareReplaceTable(xxx, xxxx, xxxx, xx);

}

...

2: Check the table exists or not. If the table doesn't exist, throw TableException(String.format("The table %s to be replaced doesn't exist. You may want to use CREATE TABLE AS statement or CREATE OR REPLACE TABLE AS statement.",tableIdentifier)).

3: Call method  Catalog#prepareReplaceTable  to try to get Tow. If, it means the repla will be done by the TwoPhaseCommitCatalogTable .

    a: If get empty, then the atomicity can not be guaranted. Flink will do the operations for Replace Table one by one without atomicity guarantee. More exactly4: If , it means which means the automic can not be guarrantten. Flink will apply the Replace automici. More extacly, it will drop the old table, create the new table, insert data into the new tables.Note: these three steps are not done in auctom,

    b: Otherwise, it expects the guarantee will be ensured to exteran connector implemter. The Flink will call method #begainTransanction before the insert job start, call method #commit after the job finish, call method #abort if the job fail or canceled.

For CREATE OR REPLACE TABLE AS  stament,  when the table exists, it'll consider it as REPLACE TABLE AS  statement. Otherwise, it'll consider it as CREATE TABLE AS  statement. 

...