[BUG] Row-writer INSERT OVERWRITE TABLE deletes and re-initializes the table instead of committing a replacecommit #19771

Bug Description

What happened:

Spark SQL INSERT OVERWRITE TABLE on the row-writer bulk-insert path (hoodie.spark.sql.insert.into.operation=bulk_insert, or the legacy hoodie.sql.bulk.insert.enable=true) deletes the whole table directory and re-initializes the table instead of committing a replacecommit. The timeline, every earlier commit, the metadata table and any pending table-service plan are gone after the statement. It logs:

WARN  org.apache.hudi.HoodieSparkSqlWriterInternal - hoodie table at <path> already exists. Deleting existing data & overwriting with new data.

followed by a fresh Loaded instants upto: Optional.empty and a metadata-table bootstrap (00000000000000000__deltacommit) inside the overwrite.

This applies to every whole-table overwrite: unpartitioned tables, and partitioned tables in static mode without a PARTITION clause. Partition-level overwrites (INSERT_OVERWRITE) are not affected because they run under SaveMode.Append.

To reproduce (Spark SQL, COW or MOR):

set hoodie.spark.sql.insert.into.operation=bulk_insert;
create table t (id int, name string, price double) using hudi tblproperties (primaryKey = 'id') location '/tmp/t';
insert into t values (1, 'a1', 10);
insert into t values (2, 'a2', 20);
call run_clustering(table => 't', op => 'schedule');          -- 1 pending clustering instant
insert overwrite table t values (3, 'b1', 30);                -- succeeds; table now has ONE commit, no pending clustering, no history

Without the bulk_insert setting the same statement commits an INSERT_OVERWRITE_TABLE replacecommit and keeps the history.

Cause:

HoodieSparkSqlWriter.handleSaveModes exempts INSERT_OVERWRITE_TABLE from the destructive SaveMode.Overwrite branch precisely so that SQL whole-table overwrites do not delete the base path:

} else if (mode == SaveMode.Overwrite && tableExists && operation != WriteOperationType.INSERT_OVERWRITE_TABLE) {
  // TODO HUDI-6286 should not delete old data if using `Overwrite` mode
  log.warn(s"hoodie table at $tablePath already exists. Deleting existing data & overwriting with new data.")
  fs.delete(tablePath, true)

But ProvidesHoodieConfig.buildHoodieInsertConfig rewrites the operation for the row-writer path: operation = BULK_INSERT with the real intent carried in HoodieInternalConfig.BULKINSERT_OVERWRITE_OPERATION_TYPE = INSERT_OVERWRITE_TABLE (ProvidesHoodieConfig.scala:244-248). deduceOperation therefore returns BULK_INSERT, the exemption does not match, and the table is deleted before DatasetBulkInsertOverwriteTableCommitActionExecutor ever runs. The executor then writes into an empty table, which is why it commits numReplaceFileIds:0.

Both halves came in together in #8076 (606bd7b17132, 2023-06-29); the exemption predates it and never learned about the rewritten operation. #15984 is the umbrella ask for Overwrite mode to stop deleting data at all; this is the narrower case where the code already intends to keep the data and a one-line guard mismatch defeats it (checking BULKINSERT_OVERWRITE_OPERATION_TYPE == INSERT_OVERWRITE_TABLE alongside operation in the condition is sufficient).

Consequences beyond the data:

• The pending-clustering guard added by fix(spark): reject INSERT_OVERWRITE when overlapping with pending clustering #18829 (BaseDatasetBulkInsertCommitActionExecutor.rejectIfOverlappingPendingClustering) can never fire for INSERT_OVERWRITE_TABLE on this path: the plan is deleted with the table before the check runs.
• TestInsertTable2 "Test bulk insert with insert overwrite table" and the whole-table half of "Test bulk insert with insert overwrite partition" pass for the wrong reason: the old rows are gone because the table was recreated, and the INSERT_OVERWRITE_TABLE operation-type assertion holds on the recreated table's single commit. The deletion warning fires in both (cow and mor).

Environment

Reproduced at d291efccaad2; handleSaveModes and the ProvidesHoodieConfig rewrite are unchanged on current master (18ae8c349058). Spark 3.5 / Scala 2.12 profile, local filesystem, metadata table enabled. Found while adding row-writer pending-clustering coverage on #19163.

Logs and Stack Trace

Sequence from the run (single test, unpartitioned COW table, two inserts then run_clustering schedule, then insert overwrite table):

Scheduling clustering at instant time: 20260827094151003 ...
Loaded instants upto: Option{val=[==>20260827094151003__clustering__REQUESTED]}
...
WARN  HoodieSparkSqlWriterInternal - hoodie table at .../htestinserttable2_17 already exists. Deleting existing data & overwriting with new data.
Loading Active commit timeline for .../htestinserttable2_17
Loaded instants upto: Optional.empty
Creating a new instant: [==>20260827094151218__replacecommit__REQUESTED]
Creating a new instant: [==>00000000000000000__deltacommit__REQUESTED]     <- metadata table bootstrapped again
Creating  metadata for INSERT_OVERWRITE_TABLE numWriteStats:1 numReplaceFileIds:0