@drmingdrmer reviewed 2 files and all commit messages, and made 1 comment.
Reviewable status: 2 of 7 files reviewed, 1 unresolved discussion (waiting on dantengsky, SkyFan2002, TCeason, and zhyass).

src/meta/api/src/api_impl/table_api.rs line 379 at r1 (raw file):

                                if !existing_meta
                                    .engine
                                    .eq_ignore_ascii_case(&req.table_meta.engine)

why is this check done only when as_dropped is set? the codes does not tell the reason behind it.

Code quote:

                            if req.as_dropped {
                                // CTAS does not call construct_drop_table_txn_operations(),
                                // so validate its existing table here.
                                let existing_meta =
                                    self.get_pb(&TableId::new(*id.data)).await?.ok_or_else(|| {
                                        KVAppError::AppError(AppError::UnknownTableId(
                                            UnknownTableId::new(
                                                *id.data,
                                                "create or replace failed to find existing table meta",
                                            ),
                                        ))
                                    })?;
                                if !existing_meta
                                    .engine
                                    .eq_ignore_ascii_case(&req.table_meta.engine)

Request changes: the normal replacement path is close, but the invariant is not enforced at the publication point for two-phase CTAS. Temporary CTAS can still publish a cross-engine replacement.

1. Temporary CTAS has a check/commit race

TempTblMgr::create_table() checks the current engine, but always returns prev_table_id: None. Later, commit_table_meta() removes the orphan and unconditionally replaces whatever currently owns the visible name.

A long-running CTAS can therefore execute this sequence:

1. Prepare hidden FUSE table while the visible table is FUSE.
2. Another query in the same session drops/recreates the visible name as MEMORY.
3. CTAS commits and silently replaces MEMORY with FUSE.

Preserve the observed ID in the prepare result:

let existing_id = self.name_to_id.get(&desc).copied();

let new_table = match (existing_id, create_option) {
    // existing branches...
};

Ok(CreateTableReply {
    // ...
    prev_table_id: as_dropped.then_some(existing_id).flatten(),
    orphan_table_name,
})

Then make the commit validate before mutating either map:

let orphan_id = self
    .name_to_id
    .get(&orphan_desc)
    .copied()
    .ok_or_else(|| ErrorCode::UnknownTable(orphan_desc.clone()))?;

let current_id = self.name_to_id.get(&desc).copied();
if current_id != req.prev_table_id {
    return Err(ErrorCode::TableVersionMismatched(format!(
        "Temporary table {desc} changed while CTAS was running"
    )));
}

if let Some(current_id) = current_id {
    let current = self.id_to_table.get(&current_id).ok_or_else(|| {
        ErrorCode::Internal(format!("Missing temporary table metadata for {current_id}"))
    })?;
    let replacement = self.id_to_table.get(&orphan_id).ok_or_else(|| {
        ErrorCode::Internal(format!("Missing temporary table metadata for {orphan_id}"))
    })?;

    TableEngineMismatch::ensure(
        &req.name_ident.table_name,
        &current.meta.engine,
        &replacement.meta.engine,
    )
    .map_err(|e| ErrorCode::from(AppError::from(e)))?;
}

// Only mutate name_to_id and id_to_table after every validation succeeds.

The important part is to perform all checks before remove(&orphan_desc), so a rejected commit preserves both tables for diagnosis and cleanup.

2. Persistent CTAS validates too early

The engine check in create_table() happens while preparing the hidden table. The as_dropped branch does not bind the observed existing TableMeta sequence to that transaction, and commit_table_meta() does not compare the previous visible table engine with the hidden table engine.

Normal replacement is protected because construct_drop_table_txn_operations() adds a condition on the preloaded metadata sequence. CTAS must get equivalent protection in the transaction that publishes the hidden table.

At commit time, load the metadata for prev_table_id, compare it with the hidden table, and add its sequence to the same transaction:

let mut replacement_meta = tb_meta.ok_or_else(|| {
    KVAppError::AppError(AppError::UnknownTableId(
        UnknownTableId::new(table_id, "commit_table_meta"),
    ))
})?;

if let Some(prev_table_id) = req.prev_table_id {
    let prev_tbid = TableId::new(prev_table_id);
    let previous = self.get_pb(&prev_tbid).await?.ok_or_else(|| {
        KVAppError::AppError(AppError::UnknownTableId(
            UnknownTableId::new(prev_table_id, "commit_table_meta previous table"),
        ))
    })?;

    TableEngineMismatch::ensure(
        req.name_ident.table_name.as_str(),
        &previous.engine,
        &replacement_meta.engine,
    )
    .map_err(|e| KVAppError::AppError(e.into()))?;

    txn_req
        .condition
        .push(txn_cond_seq(&prev_tbid, Eq, previous.seq));
}

replacement_meta.drop_on = None;

This should run after checking that the history list still ends in prev_table_id and before constructing the publish transaction. The existing name, history-list, orphan, and replacement metadata conditions should remain.

Keeping the early check is useful for fast failure, but the commit-time check must be authoritative.

3. Centralize engine compatibility

The case-insensitive comparison and error construction are duplicated in the meta API and temporary-table manager. Put the policy next to TableEngineMismatch so all paths use identical semantics:

impl TableEngineMismatch {
    pub fn ensure(
        table_name: &str,
        existing_engine: &str,
        new_engine: &str,
    ) -> std::result::Result<(), Self> {
        if existing_engine.eq_ignore_ascii_case(new_engine) {
            return Ok(());
        }

        Err(Self::new(table_name, existing_engine, new_engine))
    }
}

This also gives tests one precise domain function to exercise.

4. Do not pass unkeyed preloaded metadata

construct_drop_table_txn_operations(table_id, preloaded_table_meta, ...) relies on a comment saying the metadata must belong to table_id. The type cannot enforce that. A future mismatched call could write metadata and clean up policy or MV references using the wrong table.

Prefer either letting the drop helper load its own metadata, or bind the key and value in one private type:

struct VersionedTable {
    id: TableId,
    meta: SeqV<TableMeta>,
}

async fn construct_drop_table_txn_operations(
    kv_api: &impl KVApi,
    existing: VersionedTable,
    req: &DropTableTxnReq,
    txn: &mut TxnRequest,
) -> Result<(u64, u64), KVAppError> {
    // existing.id and existing.meta cannot be mixed independently.
}

The current helper already has ten parameters. A small request object would also make invalid boolean combinations such as if_exists plus if_delete easier to avoid.

5. Add tests for the actual two-phase operation

The current meta test stops after create_table(as_dropped = true), the stream SQL test contains no AS SELECT, and the temporary-table test covers only immediate replacement. They do not exercise CTAS commit.

Please add:

• SQL: CREATE OR REPLACE TABLE stream_name AS SELECT ... fails and preserves the stream.
• SQL or manager test: temporary cross-engine CTAS fails and preserves the visible temporary table.
• Meta API test: prepare hidden CTAS, change the previous visible metadata, then verify commit returns TableEngineMismatch without changing the name mapping, history, or either table metadata.
• Exact error assertions such as assert_eq!(actual, expected) rather than only matching the enum variant.

The design rule should be: engine compatibility is checked and transactionally guarded where the visible name becomes associated with the replacement table.

All five points from the previous round are implemented (VersionedTable, TableEngineMismatch::ensure, temp prev_table_id + commit guard, commit-time check with a seq condition, tests). But the new commit-time check introduced a regression that CI does not catch.

Blocker: prev_table_id is "last id in history", not "the table being replaced"

src/meta/api/src/api_impl/table_api.rs L1208-L1227.

In create_table, prev_table_id is tb_id_list.data.id_list.last() — the last id in the name's history list, which survives DROP. It is not "the currently visible table". So after a plain DROP, a non-replacing CTAS still carries prev_table_id = Some(<dropped table>), and the new check compares the new table against a tombstone:

CREATE STREAM t ON TABLE src;   -- t: engine STREAM, history [X]
DROP STREAM t;                  -- name is free; history still [X]
CREATE TABLE t AS SELECT 1;     -- ERROR 1302: existing table uses engine STREAM

create_table correctly skips the check here (key_dbid_tbname is absent, so there is nothing to replace), which means the failure only surfaces at commit — i.e. after the entire insert pipeline has already run. The same applies to DROP VIEW v; CREATE TABLE v AS SELECT ..., and to any ENGINE=MEMORY table dropped and then re-created via CTAS.

Verified against this branch (38ba68e) by adding a repro to SchemaApiTestSuite and running test_schema_api_single_node:

REPRO: prev_table_id = Some(6)
REPRO: commit_table_meta result = Err(AppError(TableEngineMismatch(
    TableEngineMismatch { table_name: "table", existing_engine: "STREAM", new_engine: "JSON" })))
panicked at: CTAS after DROP must succeed

The repro:

let mut util = DbTableHarness::new(mt, tenant_name, db_name, table_name, "STREAM");
util.create_db().await?;
util.create_table().await?;      // engine STREAM
util.drop_table_by_id().await?;  // name free, history still [X]

// CREATE TABLE t AS SELECT ...  (engine JSON), two-phase CTAS
let mut meta = util.table_meta();
meta.engine = "JSON".to_string();
meta.drop_on = Some(Utc::now());
let create_req = CreateTableReq {
    create_option: CreateOption::Create,
    as_dropped: true,
    table_meta: meta,
    /* ... */
};
let reply = mt.create_table(create_req.clone()).await?;
let res = mt.commit_table_meta(CommitTableMetaReq {
    name_ident: create_req.name_ident,
    db_id: reply.db_id,
    table_id: reply.table_id,
    prev_table_id: reply.prev_table_id,
    orphan_table_name: reply.orphan_table_name,
}).await;
assert!(res.is_ok());  // fails on this branch

CI is green only because every existing CTAS-after-drop test happens to keep the same engine.

Fix: gate the check on the visible name. dbid_tbname_seq is already read at L1116 and its id is discarded as _table_id — run the check only when dbid_tbname_seq != 0, and prefer comparing against that visible _table_id rather than prev_table_id, since the visible table is the one actually being replaced. The txn_cond_seq(&prev_tbid, ...) guard should move with it.

Please add a regression test: DROP an object with engine A, then CTAS with engine B under the same name, and assert success.

The temp-table path does not have this bug — TempTblMgr derives prev_table_id from name_to_id, which has no tombstones.

The temp-table engine check is unreachable, and its test does not cover it

src/query/storages/common/session/src/temp_table.rs L198-L212.

The current_id != req.prev_table_id guard three lines above already subsumes it. If the ids match, current is the same TempTable whose engine create_table already compared against table_meta, and replacement.meta is that table_meta. Temp ids are never reused and nothing mutates meta.engine, so ensure() here can never fail.

test_commit_cross_engine_ctas_preserves_temporary_tables confirms this: it asserts TableVersionMismatched, not an engine mismatch. It passes without ever executing L198-L212.

Either drop the block and rename the test to what it actually verifies (e.g. ..._preserves_temporary_tables_when_visible_table_changed), or keep it with a comment stating it is defense-in-depth. The version guard is the correct fix for the temp CTAS race.

The meta-side equivalent is also only reachable by forging KV state — table_commit_table_meta_engine_mismatch has to upsert_test_data the engine to trigger it. I would keep that one, since update_table_meta writes a whole TableMeta and the meta API cannot assume engine immutability from an arbitrary client, but please say so in a comment.

Smaller items

• Unrelated change: src/query/storages/paimon/Cargo.toml and Cargo.lock drop databend-common-storage and log. They are genuinely unused, but this is unrelated to the fix — please split it out.
• schema_api_test_suite.rs L313: assert_eq!("json", replaced_meta.engine) is vacuous. create_table_with returns the meta the test constructed, not what was stored; the only real check is that the call did not error. Assert on util.get_table_by_name(table).await?.meta.engine to verify the round trip.
• src/query/ee/src/stream/handler.rs: a rejected CREATE OR REPLACE STREAM target ON TABLE source still enables change_tracking on source beforehand. Pre-existing ordering, but it is a counterexample to the issue's "a rejected statement must leave metadata unchanged". Not blocking.
• Point 4 is partially addressed: VersionedTable removes the unkeyed-metadata hazard, but the helper still takes nine positional parameters. Fine to defer.
• ErrorCode::TableEngineNotSupported (1302) reads a little oddly for "engines differ", but it matches existing practice (interpreter_table_drop.rs L95, stream/handler.rs L189 use it for wrong-object-type DDL). No objection.