/**
 * Enables different staged operations to ensure atomicity in a {@link DynamicTableSink}.
 *
 * <p>By default, if this interface is not implemented, indicating that atomic operations are not
 * supported, then a non-atomic implementation is used.
 */
@PublicEvolving
public interface SupportsStaging {

    /**
     * Provides a {@link StagedTable} that provided transaction abstraction. StagedTable will be

@PublicEvolving
public interface Catalog {

    /**
     * Create a {@link TwoPhaseCatalogTable} that provided transaction abstraction.
     * TwoPhaseCatalogTable will be combined with {@link JobStatusHook} to achieve atomicity
     * support in the Flink framework. Default returns empty, indicating that atomic operations are
     * not supported, then using non-atomic implementations.
     *
     * <p>The framework will make sure to call this method with fully validated {@link
     * ResolvedCatalogTable}.
     *
     * @param tablePath path of the table to be created
     * @param table the table definition
     * @param ignoreIfExists flag to specify behavior when a table or view already exists at the
     *     given path: if set to false, it throws a TableAlreadyExistException, if set to true, do
     *     nothing.
     * @paramcombined isStreamingModewith A{@link flagJobStatusHook} thatto tellsachieve ifatomicity thesupport currentin tablethe isFlink in stream mode, Differentframework. Call
     * the relevant API of modesStagedTable canwhen havethe differentJob implementationsstate ofis atomicity supportswitched.
     *
 @return {@link TwoPhaseCatalogTable} that can* be<p>This serializedmethod andwill providesbe atomic
called at the compile stage.
 *     operations*
     * @throws@param TableAlreadyExistExceptionStagingContext ifTell tableDynamicTableSink, alreadythe existsoperation andtype ignoreIfExistsof isthis falseStagedTable,
     *  @throws DatabaseNotExistException if theexpandable
 database in tablePath doesn't exist
* @return {@link StagedTable} that *can @throwsbe CatalogExceptionserialized inand caseprovides of any runti
me exceptionatomic operations
     */
    defaultStagedTable Optional<TwoPhaseCatalogTable> twoPhaseCreateTable(applyStaging(StagingContext context);

    /**
     * The context ObjectPathis tablePath,
intended to tell DynamicTableSink the type of this operation. In this  CatalogBaseTable tableway,
     * DynamicTableSink can return the corresponding implementation boolean ignoreIfExists,of StagedTable according to the
     * specific operation. More types of operations booleancan isStreamingMode)
be extended in the future.
     */
   throws TableAlreadyExistException,interface DatabaseNotExistException,StagingContext CatalogException {
        returnStagingPurpose Optional.emptygetStagingPurpose();
    }

}

/**
 * A {@link StagedTable} for atomic 

/**
 * A {@link CatalogTable} for atomic semantics using a two-phase commit protocol, combined with
 * {@link JobStatusHook} for atomic CTAS. {@link TwoPhaseCatalogTableStagedTable} will be a member
 * variable of
 * CtasJobStatusHook and can be serialized;
 *
 * <p>
 * CtasJobStatusHook#onCreated <p>CtasJobStatusHook#onCreated will call the begin method of TwoPhaseCatalogTableStagedTable;
 * CtasJobStatusHook#onFinished will call the commit method of TwoPhaseCatalogTableStagedTable;
 * CtasJobStatusHook#onFailed and CtasJobStatusHook#onCanceled will call the abort method of
 * TwoPhaseCatalogTableStagedTable;
 */
@PublicEvolving
public interface TwoPhaseCatalogTableStagedTable extends CatalogTable, Serializable {

    /**
     * This method will be called when the job is started. Similar to what it means to open a
     * transaction in a relational database; In Flink's atomic CTAS scenario, it is used to do some
     * initialization work; For example, initializing the client of the underlying service, the tmp
     * path of the underlying storage, or even call the start transaction API of the underlying
     * service, etc.
     */
    void begin();

    /**
     * This method will be called when the job is succeeds. Similar to what it means to commit the
     * transaction in a relational database; In Flink's atomic CTAS scenario, it is used to do some
     * data visibility related work; For example, moving the underlying data to the target
     * directory, writing buffer data to the underlying storage service, or even call the commit
     * transaction API of the underlying service, etc.
     */
    void commit();

    /**
     * This method will be called when the job is failed or canceled. Similar to what it means to
     * rollback the transaction in a relational database; In Flink's atomic CTAS scenario, it is
     * used to do some data cleaning; For example, delete the data in tmp directory, delete the
     * temporary data in the underlying storage service, or even call the rollback transaction API
     * of the underlying service, etc.
     */
    void abort();
}

@PublicEvolving
public class TableConfigOptions {
    @Documentation.TableOption(execMode = Documentation.ExecMode.BATCH_STREAMING)
    public static final ConfigOption<Boolean> TABLE_CTAS_ATOMICITY_ENABLED =
            key("table.ctas.atomicity-enabled")
                    .booleanType()
                    .defaultValue(false)
                    .withDescription(
                     

/**
 * This Hook is used to implement the Flink CTAS atomicity semantics, calling the corresponding API
 * of {@link TwoPhaseCatalogTable} at different stages of the job.
 */
public class CtasJobStatusHook implements JobStatusHook {

    private final TwoPhaseCatalogTable twoPhaseCatalogTable;

    public CtasJobStatusHook(TwoPhaseCatalogTable twoPhaseCatalogTable) {
       "Specifies this.twoPhaseCatalogTableif =the twoPhaseCatalogTable;
create table as select }

operation is executed atomically. @Override"
     public void onCreated(JobID jobId) {
        twoPhaseCatalogTable.begin();
    }

    @Override
    public void onFinished(JobID jobId) {
 + "By default, the operation is  twoPhaseCatalogTable.commit();
    }

    @Override
    public void onFailed(JobID jobId, Throwable throwable) {
        twoPhaseCatalogTable.abort();
    }

    @Override
    public void onCanceled(JobID jobId) {
non-atomic. The target table is created in Client side, and it will not be dropped even though the job fails or is cancelled. "
                                  + "If  twoPhaseCatalogTable.abort();
    }
}

boolean isAtomicCtas = tableConfig.get(TableConfigOptions.TABLE_CTAS_ATOMICITY_ENABLED) && dynamicTableSink instanceof SupportsStaging;

/**
 * This Hook is used to implement the Flink CTAS atomicity semantics, calling the corresponding API
 * of {@link StagedTable} at different stages of the job.
 */
public class CtasJobStatusHook implements JobStatusHook {

    private final StagedTable stagedTable;

    public CtasJobStatusHook(StagedTable stagedTable) {
        this.stagedTable = stagedTable;
    }

    @Override
    public void onCreated(JobID jobId) {
        stagedTable.begin();
    }

    @Override
    public void onFinished(JobID jobId) {
        stagedTable.commit();
    }

    @Override
    public void onFailed(JobID jobId, Throwable throwable) {
        stagedTable.abort();
    }

    @Override
    public void onCanceled(JobID jobId) {
        stagedTable.abort();
    }
}

Optional<DynamicTableSink> dynamicTableSinkOptional =
        getDynamicTableSink(
                catalogTable,
                tableIdentifier,
                createTableOperation.isTemporary(),
                catalogManager.getCatalog(catalogName)); 
if (tableConfig.get(TableConfigOptions.TABLE_CTAS_ATOMICITY_ENABLED)
        && dynamicTableSinkOptional.isPresent()
        && dynamicTableSinkOptional.get() instanceof SupportsStaging) {
    DynamicTableSink dynamicTableSink = dynamicTableSinkOptional.get();
    StagedTable stagedTable =
            ((SupportsStaging) dynamicTableSink)
                    .applyStaging(
                            new SupportsStaging.StagingContext() {
                                @Override
                                public SupportsStaging.StagingPurpose
                                        getStagingPurpose() {
                                    if (createTableOperation
                                            .isIgnoreIfExists()) {
                                        return SupportsStaging.StagingPurpose
                                                .CREATE_TABLE_AS_IF_NOT_EXISTS;
                                    }
                                    return SupportsStaging.StagingPurpose
                                            .CREATE_TABLE_AS;
                                }
                            });
    CtasJobStatusHook ctasJobStatusHook = new CtasJobStatusHook(stagedTable);
    mapOperations.add(
            ctasOperation.toStagedSinkModifyOperation(
                    createTableOperation.getTableIdentifier(),
                    catalogTable,
                    ctasCatalog,
                    dynamicTableSink));
    jobStatusHookList.add(ctasJobStatusHook);
} else {
    // execute CREATE TABLE first for non-atomic CTAS statements
    executeInternal(ctasOperation.getCreateTableOperation());
    mapOperations.add(ctasOperation.toSinkModifyOperation(catalogManager));
}

@Override
public StagedTable applyStaging(StagingContext context) {
    Table hiveTable =
            HiveTableUtil.instantiateHiveTable(
                    identifier.toObjectPath(),
                    catalogTable,
                    HiveConfUtils.create(jobConf),
                    false);

    hiveStagedTable =
            new HiveStagedTable(
                    hiveVersion,
                    new JobConfWrapper(jobConf),
                    hiveTable,
                    context.getStagingPurpose()
                            == SupportsStaging.StagingPurpose.CREATE_TABLE_AS_IF_NOT_EXISTS);

    return hiveStagedTable;
}

/** An implementation of {@link StagedTable} for Hive to support atomic ctas. */
public class HiveStagedTable implements StagedTable {

    private static final long serialVersionUID = 1L;

    @Nullable private final String hiveVersion;
    private final JobConfWrapper jobConfWrapper;

    private final Table table;

    private final boolean ignoreIfExists;

    private transient HiveMetastoreClientWrapper client;

    private FileSystemFactory fsFactory;
    private TableMetaStoreFactory msFactory;
    private boolean overwrite;
    private Path tmpPath;
    private String[] partitionColumns;
    private boolean dynamicGrouped;
    private LinkedHashMap<String, String> staticPartitions;
    private ObjectIdentifier identifier;
    private PartitionCommitPolicyFactory partitionCommitPolicyFactory;

    public HiveStagedTable(
            String hiveVersion,
            JobConfWrapper jobConfWrapper,
            Table table,
            boolean ignoreIfExists) {
        this.hiveVersion = hiveVersion;
        this.jobConfWrapper = jobConfWrapper;
        this.table = table;
        this.ignoreIfExists = ignoreIfExists;
    }

    @Override
    public void begin() {
        // init hive metastore client
        client =
                HiveMetastoreClientFactory.create(
                        HiveConfUtils.create(jobConfWrapper.conf()), hiveVersion);
    }

    @Override
    public void commit() {
        try {
            // create table first
            client.createTable(table);

            try {
                List<PartitionCommitPolicy> policies = Collections.emptyList();
                if (partitionCommitPolicyFactory != null) {
                    policies =
                            partitionCommitPolicyFactory.createPolicyChain(
                                    Thread.currentThread().getContextClassLoader(),
                                    () -> {
                                        try {
                                            return fsFactory.create(tmpPath.toUri());
                                        } catch (IOException e) {
                                            throw new RuntimeException(e);
                                        }
                                    });
                }

                FileSystemCommitter committer =
                        new FileSystemCommitter(
                                fsFactory,
                                msFactory,
                                overwrite,
                                tmpPath,
                                partitionColumns.length,
                                false,
                   

	@Override
    public Optional<TwoPhaseCatalogTable> twoPhaseCreateTable(
            ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists, boolean isStreamingMode)
identifier,
                              throws TableAlreadyExistException, DatabaseNotExistExceptionstaticPartitions,
 CatalogException {

        if (isStreamingMode) {
            //HiveCatalog does not support atomicity semantics in stream mode
policies);
               return Optionalcommitter.emptycommitPartitions();
        }
		... ...
   } catch (Exception e)  TwoPhaseCatalogTable{
 twoPhaseCatalogTable = new HiveTwoPhaseCatalogTable(
            throw new   getHiveVersion(),
       TableException("Exception in two phase commit", e);
         new JobConfWrapper(JobConfUtils.createJobConfWithCredentials(hiveConf)),
  } finally {
            hiveTable,
    try {
           ignoreIfExists);

        return Optional.of(twoPhaseCatalogTable fsFactory.create(tmpPath.toUri()).delete(tmpPath, true);
    }