From 85c4ae452e17feccaa9b297fdf2a2a11a69d5b9a Mon Sep 17 00:00:00 2001 From: yujun Date: Wed, 30 Sep 2026 11:08:09 +0800 Subject: [PATCH 1/9] [fix](fe) Report a cancelled insert overwrite instead of a success that lost rows An INSERT OVERWRITE commits its rows into temporary partitions and publishes them with a swap afterwards, so a cancellation that lands between the two halves is too late to take anything back: the rows are durable, and everything the read consumed -- the base table stream offsets among it -- was committed with that same transaction. The window was answered by dropping the temporary partitions and returning normally, which published nothing while telling the client the overwrite had succeeded, and left the consumed offsets advanced past rows no target will ever hold (a re-run reads from the advanced offset and silently misses that range). The cancellations that land before the rows are committed were answered the same way, reporting success for an overwrite that never ran. Key changes: - A cancellation that lands before the rows are committed now fails the statement, like the cancellation the inner insert already reports for itself. Nothing durable happened, so the offsets are where they were and a re-run reads the same rows. - A cancellation that lands after the rows are committed completes the swap instead: the rows the read consumed are the ones the table holds, the offset advance and the publication stay aligned, and the success the statement reports is the outcome it is. - The failure path no longer asks the overwrite manager to fail the task of a cancellation that landed before any task was registered. - A debug point (stage=beforeTheInsert|afterTheInsert, scoped by table name) injects the cancellation at either side of the insert. Unit Test: - regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy covers both sides: a cancelled statement leaves the table untouched, and a cancellation inside the window is published rather than dropped. - insert_overwrite_p0 (14 suites, including the new one) and mtmv_p0/ivm/test_ivm_overwrite_failure_between_the_halves pass. --- .../insert/InsertOverwriteTableCommand.java | 88 ++++++++++++--- .../test_insert_overwrite_cancel.out | 9 ++ .../test_insert_overwrite_cancel.groovy | 102 ++++++++++++++++++ 3 files changed, 182 insertions(+), 17 deletions(-) create mode 100644 regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out create mode 100644 regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java index 6557553bc2306e..b6f497a35544a1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java @@ -111,6 +111,23 @@ public class InsertOverwriteTableCommand extends Command public static final String DEBUG_POINT_FAIL_BETWEEN_THE_HALVES_OF_AN_OVERWRITE = "InsertOverwriteTableCommand.failBetweenTheTwoHalvesOfAnOverwrite"; + /** + * Cancels the overwrite at a chosen point of the command, so that both halves of what a cancellation means + * can be pinned by a test: one that lands before the rows are committed takes the statement back and the + * statement has to fail, one that lands after them cannot take anything back and the overwrite has to + * complete. See test_insert_overwrite_cancel. + * + *

The point's {@code stage} parameter is {@code beforeTheInsert} or {@code afterTheInsert}, and its + * {@code table_name} parameter names the one table the point may disturb: the read below answers with the + * default when the point is not enabled or carries no such parameter, and no table is named by an empty + * string, so an enabled point cannot disturb an overwrite that is not the one under test. + */ + public static final String DEBUG_POINT_CANCEL_AN_OVERWRITE = + "InsertOverwriteTableCommand.cancelAnOverwrite"; + + private static final String STAGE_BEFORE_THE_INSERT = "beforeTheInsert"; + private static final String STAGE_AFTER_THE_INSERT = "afterTheInsert"; + private static final Logger LOG = LogManager.getLogger(InsertOverwriteTableCommand.class); private LogicalPlan originLogicalQuery; @@ -266,32 +283,37 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { } else { // it's overwrite table(as all partitions) or specific partition(s) List tempPartitionNames = InsertOverwriteUtil.generateTempPartitionNames(partitionNames); + cancelTheOverwriteAt(STAGE_BEFORE_THE_INSERT, targetTable); if (isCancelled.get()) { - LOG.info("insert overwrite is cancelled before registerTask, queryId: {}", - ctx.getQueryIdentifier()); - return; + // Nothing durable happened: no task is registered, no temp partition exists, no row was + // written and nothing was committed. The statement is a plain failure, like the one the + // inner insert reports when it is cancelled, rather than the success of an overwrite that + // did not run. + throw cancelledBeforeTheRowsWereCommitted("before registerTask", ctx); } taskId = insertOverwriteManager.registerTask(targetTable, tempPartitionNames); if (isCancelled.get()) { - LOG.info("insert overwrite is cancelled before addTempPartitions, queryId: {}", - ctx.getQueryIdentifier()); - // not need deal temp partition - insertOverwriteManager.taskSuccess(taskId); - return; + // The catch below takes the registration back; no temp partition exists yet, so there is + // nothing else to drop. + throw cancelledBeforeTheRowsWereCommitted("before addTempPartitions", ctx); } InsertOverwriteUtil.addTempPartitions(targetTable, partitionNames, tempPartitionNames); if (isCancelled.get()) { - LOG.info("insert overwrite is cancelled before insertInto, queryId: {}", ctx.getQueryIdentifier()); - insertOverwriteManager.taskFail(taskId); - return; + // The catch below drops the temp partitions this cancelled statement created. + throw cancelledBeforeTheRowsWereCommitted("before insertInto", ctx); } // todo: need to refresh remote target table after add temp partitions insertIntoPartitions(ctx, executor, tempPartitionNames, wholeTable); + cancelTheOverwriteAt(STAGE_AFTER_THE_INSERT, targetTable); if (isCancelled.get()) { - LOG.info("insert overwrite is cancelled before replacePartition, queryId: {}", - ctx.getQueryIdentifier()); - insertOverwriteManager.taskFail(taskId); - return; + // Too late to cancel: insertIntoPartitions returns only once its transaction has committed + // and published the rows into the temp partitions, and everything the read consumed -- the + // base table stream offsets among it -- was committed with that same transaction. Dropping + // the temp partitions here is exactly what would lose those rows against an advanced + // offset, while the swap below is what publishes them. The overwrite completes, and it is + // the outcome the statement reports. + LOG.info("insert overwrite is cancelled after its rows were committed, completing it," + + " queryId: {}", ctx.getQueryIdentifier()); } failBetweenTheTwoHalvesOfAnOverwrite(targetTable); InsertOverwriteUtil.replacePartition(targetTable, partitionNames, tempPartitionNames, @@ -304,9 +326,10 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { } } catch (Exception e) { LOG.warn("insert into overwrite failed with task(or group) id {}", taskId, e); - if (isAutoDetectOverwrite(getLogicalQuery())) { + // A cancel that landed before registerTask leaves nothing registered to fail, and no id was taken. + if (isAutoDetectOverwrite(getLogicalQuery()) && taskId != 0) { insertOverwriteManager.taskGroupFail(taskId); - } else { + } else if (taskId != 0) { insertOverwriteManager.taskFail(taskId); } throw e; @@ -386,6 +409,37 @@ private static void failBetweenTheTwoHalvesOfAnOverwrite(TableIf targetTable) th throw new UserException("debug point: " + DEBUG_POINT_FAIL_BETWEEN_THE_HALVES_OF_AN_OVERWRITE); } + /** + * Cancels this overwrite when the debug point names the stage and the table it targets; see the constant + * above. Nothing here decides what a cancelled overwrite means -- the two call sites do, and they differ: + * one takes the statement back, the other cannot. + */ + private void cancelTheOverwriteAt(String stage, TableIf targetTable) { + if (!stage.equals(DebugPointUtil.getDebugParamOrDefault( + DEBUG_POINT_CANCEL_AN_OVERWRITE, "stage", ""))) { + return; + } + if (!targetTable.getName().equals(DebugPointUtil.getDebugParamOrDefault( + DEBUG_POINT_CANCEL_AN_OVERWRITE, "table_name", ""))) { + return; + } + LOG.info("debug point {} cancels the overwrite of {} at {}", DEBUG_POINT_CANCEL_AN_OVERWRITE, + targetTable.getName(), stage); + cancel(); + } + + /** + * The failure a cancellation that landed before anything durable is reported as. The caller wrote and + * committed nothing, and the offsets of the streams it was about to read are where they were, so a re-run + * reads the same rows -- which is why this is a failure rather than the success of an overwrite that never + * ran. The other half of the decision, a cancellation that lands after the rows are committed, is taken + * where the swap runs. + */ + private static UserException cancelledBeforeTheRowsWereCommitted(String stage, ConnectContext ctx) { + return new UserException("insert overwrite is cancelled " + stage + ", queryId: " + + ctx.getQueryIdentifier()); + } + private void runInsertCommand(LogicalPlan logicalQuery, InsertCommandContext insertCtx, ConnectContext ctx, StmtExecutor executor) throws Exception { InsertIntoTableCommand insertCommand = new InsertIntoTableCommand(logicalQuery, labelName, diff --git a/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out new file mode 100644 index 00000000000000..dd7c2b32c975fb --- /dev/null +++ b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out @@ -0,0 +1,9 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !dst_after_the_cancelled_statement -- +1 2026-01-10 100 +2 2026-02-10 200 + +-- !dst_after_the_cancelled_window -- +3 2026-01-20 300 +4 2026-02-20 400 + diff --git a/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy new file mode 100644 index 00000000000000..24c2b455157988 --- /dev/null +++ b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy @@ -0,0 +1,102 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +/** + * What a cancelled overwrite owes the rows it has already committed, and what it owes the caller when it has + * not committed anything yet. + * + *

An overwrite commits its rows into temporary partitions and publishes them with a swap afterwards, so a + * cancellation that lands between the two halves cannot take the rows back: they are durable, and everything + * the write read -- the base table stream offsets among it -- was committed with them. Dropping the temporary + * partitions there, and answering the client with a success for an overwrite that never happened, is what + * loses those rows against an advanced offset. What the command has to do instead is finish the swap. A + * cancellation that lands before the first half has committed anything has nothing to take back, and there + * the statement has to fail rather than report the success of an overwrite that did not run. + * + *

Both are injected by one debug point in the overwrite, scoped by stage and by table name, so each case + * lands on a real statement rather than on a race. + */ +suite("test_insert_overwrite_cancel", "nonConcurrent") { + def cancelPoint = "InsertOverwriteTableCommand.cancelAnOverwrite" + + GetDebugPoint().disableDebugPointForAllFEs(cancelPoint) + sql """DROP TABLE IF EXISTS test_iot_cancel_src""" + sql """DROP TABLE IF EXISTS test_iot_cancel_dst""" + + sql """ + CREATE TABLE test_iot_cancel_src ( + id BIGINT NOT NULL, + dt DATE NOT NULL, + amount INT + ) ENGINE = OLAP + UNIQUE KEY(id, dt) + PARTITION BY RANGE(dt) ( + PARTITION p1 VALUES [('2026-01-01'), ('2026-02-01')), + PARTITION p2 VALUES [('2026-02-01'), ('2026-03-01')) + ) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "true" + ) + """ + sql """ + CREATE TABLE test_iot_cancel_dst ( + id BIGINT NOT NULL, + dt DATE NOT NULL, + amount INT + ) ENGINE = OLAP + UNIQUE KEY(id, dt) + PARTITION BY RANGE(dt) ( + PARTITION p1 VALUES [('2026-01-01'), ('2026-02-01')), + PARTITION p2 VALUES [('2026-02-01'), ('2026-03-01')) + ) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "true" + ) + """ + sql """INSERT INTO test_iot_cancel_dst VALUES (1, '2026-01-10', 100), (2, '2026-02-10', 200)""" + sql """INSERT INTO test_iot_cancel_src VALUES (3, '2026-01-20', 300), (4, '2026-02-20', 400)""" + + // A cancellation that lands before the rows are committed has nothing to take back, so the statement has + // to report it, and the table has to be where it was. + try { + GetDebugPoint().enableDebugPointForAllFEs(cancelPoint, + [stage: "beforeTheInsert", table_name: "test_iot_cancel_dst"]) + test { + sql """INSERT OVERWRITE TABLE test_iot_cancel_dst SELECT * FROM test_iot_cancel_src""" + exception "insert overwrite is cancelled before registerTask" + } + } finally { + GetDebugPoint().disableDebugPointForAllFEs(cancelPoint) + } + order_qt_dst_after_the_cancelled_statement """SELECT id, dt, amount FROM test_iot_cancel_dst""" + + // A cancellation that lands after the rows are committed cannot take them back: they are in the temporary + // partitions, and the swap is what publishes them. The statement reports the success it now is, and the + // rows it read are the ones the table holds. + try { + GetDebugPoint().enableDebugPointForAllFEs(cancelPoint, + [stage: "afterTheInsert", table_name: "test_iot_cancel_dst"]) + sql """INSERT OVERWRITE TABLE test_iot_cancel_dst SELECT * FROM test_iot_cancel_src""" + } finally { + GetDebugPoint().disableDebugPointForAllFEs(cancelPoint) + } + order_qt_dst_after_the_cancelled_window """SELECT id, dt, amount FROM test_iot_cancel_dst""" +} From 215bcf07bd59d9981ed5549ee53b61e39083b232 Mon Sep 17 00:00:00 2001 From: yujun Date: Wed, 30 Sep 2026 12:47:02 +0800 Subject: [PATCH 2/9] [fix](fe) Honour a cancelled overwrite whose insert committed nothing The window between an overwrite's two halves treated every cancellation that landed in it as too late, and completed the swap. That is right where the insert committed the rows it read, but an insert whose plan folded to an empty relation takes the path that begins no transaction at all: it commits nothing, so completing the swap published an empty table for a statement the client had cancelled, and reported success for it. The auto-detect route left the same boundary unhandled, and the two cancel points shared one debug point and two lookups, which let a point armed with execute=1 be spent by the other site's lookup and silently not fire. Key changes: - The insert marks the context it ran under when it takes the path that begins no transaction, and the overwrite reads that back: a cancellation landing in the window now fails the statement and drops the empty temporary partitions, while one that lands after committed rows still completes the swap. - The auto-detect route takes the same decision. - One debug point per cancel site, each reading its single table_name parameter from one lookup. - The suite covers the empty-plan window as well: the cancelled statement has to fail and the table has to keep its rows, which also pins that the injection reached the window. Unit Test: - regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy - insert_overwrite_p0 (14 suites) and mtmv_p0/ivm/test_ivm_overwrite_failure_between_the_halves pass. --- .../commands/insert/InsertCommandContext.java | 17 ++++ .../insert/InsertIntoTableCommand.java | 4 + .../insert/InsertOverwriteTableCommand.java | 97 ++++++++++++------- .../test_insert_overwrite_cancel.out | 3 + .../test_insert_overwrite_cancel.groovy | 86 +++++++++++++--- 5 files changed, 157 insertions(+), 50 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertCommandContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertCommandContext.java index 1b4e28117bc06e..967f0ebb54898b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertCommandContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertCommandContext.java @@ -23,4 +23,21 @@ */ public abstract class InsertCommandContext { + /** + * Set when the insert that ran under this context finished on the path that begins no transaction at + * all: its plan folded to an empty relation and there was no table stream offset to commit. Nothing + * such an insert did is durable -- no row, no offset -- which is what a caller that owns what happens + * next has to know before it decides whether a cancellation arriving from there on still has anything + * to take back. See {@link InsertIntoTableCommand#runInternal} for the one path that sets it and + * {@code InsertOverwriteTableCommand#run} for the caller that reads it. + */ + private boolean committedNothing = false; + + public boolean hasCommittedNothing() { + return committedNothing; + } + + public void setCommittedNothing(boolean committedNothing) { + this.committedNothing = committedNothing; + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java index 04fea4794ce964..065b6b03db1767 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java @@ -756,6 +756,10 @@ private void runInternal(ConnectContext ctx, StmtExecutor executor) throws Excep AbstractInsertExecutor insertExecutor = initPlan(ctx, executor); // An empty Table Stream read still needs to commit its offset update atomically. if (!insertExecutor.requiresTransaction()) { + // Nothing was committed here and nothing will be, so a caller that treats a cancellation + // arriving after this insert as too late has nothing to keep: it has to know that the rows it + // was going to publish do not exist. See InsertCommandContext#setCommittedNothing. + insertCtx.ifPresent(insertCommandContext -> insertCommandContext.setCommittedNothing(true)); return; } if (insertExecutorListener != null) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java index b6f497a35544a1..402f3a42371188 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java @@ -112,21 +112,29 @@ public class InsertOverwriteTableCommand extends Command "InsertOverwriteTableCommand.failBetweenTheTwoHalvesOfAnOverwrite"; /** - * Cancels the overwrite at a chosen point of the command, so that both halves of what a cancellation means - * can be pinned by a test: one that lands before the rows are committed takes the statement back and the - * statement has to fail, one that lands after them cannot take anything back and the overwrite has to - * complete. See test_insert_overwrite_cancel. + * Cancels the overwrite before it has committed anything, so that the half of a cancellation's meaning + * that takes the statement back can be pinned by a test: nothing durable happened, so the statement has + * to fail rather than report the success of an overwrite that did not run. See test_insert_overwrite_cancel. * - *

The point's {@code stage} parameter is {@code beforeTheInsert} or {@code afterTheInsert}, and its - * {@code table_name} parameter names the one table the point may disturb: the read below answers with the - * default when the point is not enabled or carries no such parameter, and no table is named by an empty - * string, so an enabled point cannot disturb an overwrite that is not the one under test. + *

Its {@code table_name} parameter names the one table the point may disturb: the read below answers + * with the default when the point is not enabled or carries no such parameter, and no table is named by an + * empty string, so an enabled point cannot disturb an overwrite that is not the one under test. */ - public static final String DEBUG_POINT_CANCEL_AN_OVERWRITE = - "InsertOverwriteTableCommand.cancelAnOverwrite"; + public static final String DEBUG_POINT_CANCEL_BEFORE_THE_INSERT_OF_AN_OVERWRITE = + "InsertOverwriteTableCommand.cancelBeforeTheInsertOfAnOverwrite"; - private static final String STAGE_BEFORE_THE_INSERT = "beforeTheInsert"; - private static final String STAGE_AFTER_THE_INSERT = "afterTheInsert"; + /** + * Cancels the overwrite in the window between its two halves -- after the insert, before the swap -- so + * that the other half of a cancellation's meaning can be pinned: where the rows are durable, the swap runs + * and the statement reports the success it is; where the insert committed nothing, the cancellation still + * has everything to take back. See test_insert_overwrite_cancel. + * + *

Separate from the point above rather than one point with a stage parameter: a point is consumed by + * the first lookup that reads it (see {@code DebugPointUtil#getDebugPoint}), so a shared name would let + * the check at one site spend the other site's allowance and make an armed point silently not fire. + */ + public static final String DEBUG_POINT_CANCEL_BETWEEN_THE_HALVES_OF_AN_OVERWRITE = + "InsertOverwriteTableCommand.cancelBetweenTheTwoHalvesOfAnOverwrite"; private static final Logger LOG = LogManager.getLogger(InsertOverwriteTableCommand.class); @@ -278,12 +286,18 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { // When inserting, BE will call to replace partition by FrontendService. FE will register new temp // partitions and return. for transactional, the replacement will really occur when insert successed, // i.e. `insertInto` finished. then we call taskGroupSuccess to make replacement. - insertIntoAutoDetect(ctx, executor, taskId); + InsertCommandContext insertCtx = insertIntoAutoDetect(ctx, executor, taskId); + if (isCancelled.get() && insertCtx.hasCommittedNothing()) { + // The load committed no row, so the cancellation still has everything to take back: the + // catch drops the group's temp partitions (there are none), and the statement fails + // rather than publishing a replacement the client cancelled. + throw cancelledBeforeTheRowsWereCommitted("after a load that committed nothing", ctx); + } insertOverwriteManager.taskGroupSuccess(taskId, (OlapTable) targetTable, isForceDropPartition()); } else { // it's overwrite table(as all partitions) or specific partition(s) List tempPartitionNames = InsertOverwriteUtil.generateTempPartitionNames(partitionNames); - cancelTheOverwriteAt(STAGE_BEFORE_THE_INSERT, targetTable); + cancelTheOverwriteAt(DEBUG_POINT_CANCEL_BEFORE_THE_INSERT_OF_AN_OVERWRITE, targetTable); if (isCancelled.get()) { // Nothing durable happened: no task is registered, no temp partition exists, no row was // written and nothing was committed. The statement is a plain failure, like the one the @@ -303,9 +317,17 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { throw cancelledBeforeTheRowsWereCommitted("before insertInto", ctx); } // todo: need to refresh remote target table after add temp partitions - insertIntoPartitions(ctx, executor, tempPartitionNames, wholeTable); - cancelTheOverwriteAt(STAGE_AFTER_THE_INSERT, targetTable); + InsertCommandContext insertCtx = insertIntoPartitions(ctx, executor, tempPartitionNames, wholeTable); + cancelTheOverwriteAt(DEBUG_POINT_CANCEL_BETWEEN_THE_HALVES_OF_AN_OVERWRITE, targetTable); if (isCancelled.get()) { + if (insertCtx.hasCommittedNothing()) { + // The insert took the path that begins no transaction -- its plan folded to an empty + // relation and there was no stream offset to commit -- so this window holds no durable + // work at all. Completing the swap would publish an empty table for a statement the + // client cancelled; the catch drops the empty temp partitions instead and the + // statement fails, which is the same boundary the cancellations above sit on. + throw cancelledBeforeTheRowsWereCommitted("after an insert that committed nothing", ctx); + } // Too late to cancel: insertIntoPartitions returns only once its transaction has committed // and published the rows into the temp partitions, and everything the read consumed -- the // base table stream offsets among it -- was committed with that same transaction. Dropping @@ -410,30 +432,28 @@ private static void failBetweenTheTwoHalvesOfAnOverwrite(TableIf targetTable) th } /** - * Cancels this overwrite when the debug point names the stage and the table it targets; see the constant - * above. Nothing here decides what a cancelled overwrite means -- the two call sites do, and they differ: - * one takes the statement back, the other cannot. + * Cancels this overwrite when the debug point names the table it targets; see the constants above. + * Nothing here decides what a cancelled overwrite means -- the call sites do, and they differ: the ones + * before the rows are durable take the statement back, the one after them does not. + * + *

One lookup, because a point is consumed by the lookup that reads it: reading it twice with + * {@code execute=1} armed would have the first read spend the allowance and the point be gone before the + * second, which would silently leave the overwrite uncancelled. */ - private void cancelTheOverwriteAt(String stage, TableIf targetTable) { - if (!stage.equals(DebugPointUtil.getDebugParamOrDefault( - DEBUG_POINT_CANCEL_AN_OVERWRITE, "stage", ""))) { - return; - } + private void cancelTheOverwriteAt(String debugPointName, TableIf targetTable) { if (!targetTable.getName().equals(DebugPointUtil.getDebugParamOrDefault( - DEBUG_POINT_CANCEL_AN_OVERWRITE, "table_name", ""))) { + debugPointName, "table_name", ""))) { return; } - LOG.info("debug point {} cancels the overwrite of {} at {}", DEBUG_POINT_CANCEL_AN_OVERWRITE, - targetTable.getName(), stage); + LOG.info("debug point {} cancels the overwrite of {}", debugPointName, targetTable.getName()); cancel(); } /** - * The failure a cancellation that landed before anything durable is reported as. The caller wrote and - * committed nothing, and the offsets of the streams it was about to read are where they were, so a re-run - * reads the same rows -- which is why this is a failure rather than the success of an overwrite that never - * ran. The other half of the decision, a cancellation that lands after the rows are committed, is taken - * where the swap runs. + * The failure a cancellation that found nothing durable is reported as. No row and no stream offset was + * committed, so a re-run reads the same rows -- which is why this is a failure rather than the success of + * an overwrite that never ran. What a cancellation means on the other side of that boundary, where the + * rows are durable, is decided where the swap runs. */ private static UserException cancelledBeforeTheRowsWereCommitted(String stage, ConnectContext ctx) { return new UserException("insert overwrite is cancelled " + stage + ", queryId: " @@ -459,9 +479,11 @@ private void runInsertCommand(LogicalPlan logicalQuery, InsertCommandContext ins * @param executor executor * @param tempPartitionNames tempPartitionNames * @param wholeTable overwrite target is the whole table. not one by one by partitions(...) + * @return the context the inner insert ran under, which says whether it committed anything; see + * {@link InsertCommandContext#hasCommittedNothing()} */ - private void insertIntoPartitions(ConnectContext ctx, StmtExecutor executor, List tempPartitionNames, - boolean wholeTable) + private InsertCommandContext insertIntoPartitions(ConnectContext ctx, StmtExecutor executor, + List tempPartitionNames, boolean wholeTable) throws Exception { // copy sink tot replace by tempPartitions UnboundLogicalSink copySink; @@ -512,6 +534,7 @@ private void insertIntoPartitions(ConnectContext ctx, StmtExecutor executor, Lis throw new UserException("Current catalog does not support insert overwrite yet."); } runInsertCommand(copySink, insertCtx, ctx, executor); + return insertCtx; } /** @@ -519,8 +542,11 @@ private void insertIntoPartitions(ConnectContext ctx, StmtExecutor executor, Lis * * @param ctx ctx * @param executor executor + * @return the context the inner insert ran under, which says whether it committed anything; see + * {@link InsertCommandContext#hasCommittedNothing()} */ - private void insertIntoAutoDetect(ConnectContext ctx, StmtExecutor executor, long groupId) throws Exception { + private InsertCommandContext insertIntoAutoDetect(ConnectContext ctx, StmtExecutor executor, long groupId) + throws Exception { InsertCommandContext insertCtx; LogicalPlan logicalQuery = getLogicalQuery(); if (logicalQuery instanceof UnboundTableSink) { @@ -533,6 +559,7 @@ private void insertIntoAutoDetect(ConnectContext ctx, StmtExecutor executor, lon throw new UserException("Current catalog does not support insert overwrite with auto-detect partition."); } runInsertCommand(logicalQuery, insertCtx, ctx, executor); + return insertCtx; } @Override diff --git a/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out index dd7c2b32c975fb..1553c77f726d2f 100644 --- a/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out +++ b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out @@ -7,3 +7,6 @@ 3 2026-01-20 300 4 2026-02-20 400 +-- !dst_after_the_cancelled_empty_overwrite -- +1 2026-01-10 100 + diff --git a/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy index 24c2b455157988..6361cd98f5512b 100644 --- a/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy +++ b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy @@ -17,25 +17,31 @@ /** * What a cancelled overwrite owes the rows it has already committed, and what it owes the caller when it has - * not committed anything yet. + * committed nothing. * *

An overwrite commits its rows into temporary partitions and publishes them with a swap afterwards, so a * cancellation that lands between the two halves cannot take the rows back: they are durable, and everything * the write read -- the base table stream offsets among it -- was committed with them. Dropping the temporary * partitions there, and answering the client with a success for an overwrite that never happened, is what - * loses those rows against an advanced offset. What the command has to do instead is finish the swap. A - * cancellation that lands before the first half has committed anything has nothing to take back, and there - * the statement has to fail rather than report the success of an overwrite that did not run. + * loses those rows against an advanced offset; the swap is what has to run instead. Where the insert committed + * nothing at all -- its plan folded to an empty relation, so it took the insert's no-transaction path -- the + * cancellation still has everything to take back, and completing the swap there would publish an empty table + * for a statement the client cancelled. * - *

Both are injected by one debug point in the overwrite, scoped by stage and by table name, so each case - * lands on a real statement rather than on a race. + *

Each case is injected by its own debug point in the overwrite, scoped by table name, so it lands on a + * real statement rather than on a race. The same window is asserted twice: once for the rows it publishes, + * once for the empty plan, whose statement fails only if the injection reached the window. */ suite("test_insert_overwrite_cancel", "nonConcurrent") { - def cancelPoint = "InsertOverwriteTableCommand.cancelAnOverwrite" + def beforePoint = "InsertOverwriteTableCommand.cancelBeforeTheInsertOfAnOverwrite" + def betweenPoint = "InsertOverwriteTableCommand.cancelBetweenTheTwoHalvesOfAnOverwrite" - GetDebugPoint().disableDebugPointForAllFEs(cancelPoint) + GetDebugPoint().disableDebugPointForAllFEs(beforePoint) + GetDebugPoint().disableDebugPointForAllFEs(betweenPoint) sql """DROP TABLE IF EXISTS test_iot_cancel_src""" sql """DROP TABLE IF EXISTS test_iot_cancel_dst""" + sql """DROP TABLE IF EXISTS test_iot_cancel_flat_src""" + sql """DROP TABLE IF EXISTS test_iot_cancel_flat_dst""" sql """ CREATE TABLE test_iot_cancel_src ( @@ -74,17 +80,49 @@ suite("test_insert_overwrite_cancel", "nonConcurrent") { sql """INSERT INTO test_iot_cancel_dst VALUES (1, '2026-01-10', 100), (2, '2026-02-10', 200)""" sql """INSERT INTO test_iot_cancel_src VALUES (3, '2026-01-20', 300), (4, '2026-02-20', 400)""" - // A cancellation that lands before the rows are committed has nothing to take back, so the statement has + // Unpartitioned, for the empty-plan case below: a partitioned target gives the plan a coordinator exchange, + // so the sink's child is not the empty relation and the insert commits an empty transaction instead of + // taking the path that commits nothing. + sql """ + CREATE TABLE test_iot_cancel_flat_src ( + id BIGINT NOT NULL, + dt DATE NOT NULL, + amount INT + ) ENGINE = OLAP + UNIQUE KEY(id, dt) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "true" + ) + """ + sql """ + CREATE TABLE test_iot_cancel_flat_dst ( + id BIGINT NOT NULL, + dt DATE NOT NULL, + amount INT + ) ENGINE = OLAP + UNIQUE KEY(id, dt) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "true" + ) + """ + sql """INSERT INTO test_iot_cancel_flat_dst VALUES (1, '2026-01-10', 100)""" + sql """INSERT INTO test_iot_cancel_flat_src VALUES (3, '2026-01-20', 300)""" + + // A cancellation that lands before anything was committed has nothing to take back, so the statement has // to report it, and the table has to be where it was. try { - GetDebugPoint().enableDebugPointForAllFEs(cancelPoint, - [stage: "beforeTheInsert", table_name: "test_iot_cancel_dst"]) + GetDebugPoint().enableDebugPointForAllFEs(beforePoint, + [table_name: "test_iot_cancel_dst"]) test { sql """INSERT OVERWRITE TABLE test_iot_cancel_dst SELECT * FROM test_iot_cancel_src""" exception "insert overwrite is cancelled before registerTask" } } finally { - GetDebugPoint().disableDebugPointForAllFEs(cancelPoint) + GetDebugPoint().disableDebugPointForAllFEs(beforePoint) } order_qt_dst_after_the_cancelled_statement """SELECT id, dt, amount FROM test_iot_cancel_dst""" @@ -92,11 +130,29 @@ suite("test_insert_overwrite_cancel", "nonConcurrent") { // partitions, and the swap is what publishes them. The statement reports the success it now is, and the // rows it read are the ones the table holds. try { - GetDebugPoint().enableDebugPointForAllFEs(cancelPoint, - [stage: "afterTheInsert", table_name: "test_iot_cancel_dst"]) + GetDebugPoint().enableDebugPointForAllFEs(betweenPoint, + [table_name: "test_iot_cancel_dst"]) sql """INSERT OVERWRITE TABLE test_iot_cancel_dst SELECT * FROM test_iot_cancel_src""" } finally { - GetDebugPoint().disableDebugPointForAllFEs(cancelPoint) + GetDebugPoint().disableDebugPointForAllFEs(betweenPoint) } order_qt_dst_after_the_cancelled_window """SELECT id, dt, amount FROM test_iot_cancel_dst""" + + // The same cancellation, but the insert committed nothing: an empty plan takes the insert's + // no-transaction path, so the temporary partitions hold no rows and there is no offset to keep. The + // statement has to fail and the table has to keep the rows it had -- a swap here would publish the empty + // result for a statement the client cancelled. This case also pins that the cancellation above is really + // delivered: its assertion holds only if the injection reached the window. + try { + GetDebugPoint().enableDebugPointForAllFEs(betweenPoint, + [table_name: "test_iot_cancel_flat_dst"]) + test { + sql """INSERT OVERWRITE TABLE test_iot_cancel_flat_dst + SELECT id, dt, amount FROM test_iot_cancel_flat_src WHERE 1 = 0""" + exception "insert overwrite is cancelled after an insert that committed nothing" + } + } finally { + GetDebugPoint().disableDebugPointForAllFEs(betweenPoint) + } + order_qt_dst_after_the_cancelled_empty_overwrite """SELECT id, dt, amount FROM test_iot_cancel_flat_dst""" } From 83bac904ca8555873d587f328c33d9e3eb50f697 Mon Sep 17 00:00:00 2001 From: yujun Date: Wed, 30 Sep 2026 14:29:34 +0800 Subject: [PATCH 3/9] [fix](fe) Publish an overwrite's committed rows when the response is an error The overwrite stopped at any error state the insert left behind, and dropped the temporary partitions it had written. A load whose publication times out after its commit is committed, though, and the session's visibility-timeout mode (insert_visible_timeout_return_mode=error) reports that timeout as an error: the rows existed, the response said otherwise, and the overwrite discarded them together with the stream offsets the read had consumed. The question that decides a cancellation in the same window -- did the insert commit -- answers this one as well. Key changes: - The insert context records that the transaction committed, set by each executor where its own commit happens (OLAP, remote OLAP, external/plugin). That is a different question from the response state the insert leaves behind. - The overwrite reads it wherever it decides: a cancellation landing in the window is honoured only when nothing was committed, and an error state no longer stops the swap when the rows are durable -- the swap publishes them, and the client keeps the error the session asked for. - The suite reaches the publication timeout by blocking the publish daemon, as test_insert_visible_timeout_return_mode does, and asserts that the rows the error reported as committed become visible once publish resumes. Unit Test: - regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy - insert_overwrite_p0 (14 suites), insert_p0/test_insert_visible_timeout_return_mode and mtmv_p0/ivm/test_ivm_overwrite_failure_between_the_halves pass. --- .../insert/AbstractInsertExecutor.java | 11 ++++ .../BaseExternalTableInsertExecutor.java | 1 + .../commands/insert/InsertCommandContext.java | 26 ++++++---- .../insert/InsertIntoTableCommand.java | 8 +-- .../insert/InsertOverwriteTableCommand.java | 24 ++++++--- .../commands/insert/OlapInsertExecutor.java | 4 ++ .../insert/RemoteOlapInsertExecutor.java | 1 + .../test_insert_overwrite_cancel.out | 5 ++ .../test_insert_overwrite_cancel.groovy | 51 +++++++++++++++++-- 9 files changed, 104 insertions(+), 27 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java index bc501b5816ba04..78d6375aed382f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java @@ -355,6 +355,17 @@ public boolean requiresTransaction() { return !emptyInsert || !streamUpdateInfos.isEmpty(); } + /** + * Records on the insert context that what this insert wrote is durable: its transaction has committed, + * whether or not its publication has finished. Each executor calls this where its own commit happens, + * because the response the client gets is a different question -- a publication timeout that follows a + * commit is reported as an error while the rows are committed. See + * {@link InsertCommandContext#setCommitted}. + */ + protected void markCommitted() { + insertCtx.ifPresent(insertCommandContext -> insertCommandContext.setCommitted(true)); + } + public void setStreamUpdateInfos(List streamUpdateInfos) { this.streamUpdateInfos = streamUpdateInfos; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/BaseExternalTableInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/BaseExternalTableInsertExecutor.java index 11d9dd8ee8d608..b13027e99ff72b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/BaseExternalTableInsertExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/BaseExternalTableInsertExecutor.java @@ -122,6 +122,7 @@ protected void onComplete() throws UserException { transactionManager.commit(txnId); } txnStatus = TransactionStatus.COMMITTED; + markCommitted(); long t2 = System.currentTimeMillis(); try { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertCommandContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertCommandContext.java index 967f0ebb54898b..41dce99761f9b5 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertCommandContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertCommandContext.java @@ -24,20 +24,24 @@ public abstract class InsertCommandContext { /** - * Set when the insert that ran under this context finished on the path that begins no transaction at - * all: its plan folded to an empty relation and there was no table stream offset to commit. Nothing - * such an insert did is durable -- no row, no offset -- which is what a caller that owns what happens - * next has to know before it decides whether a cancellation arriving from there on still has anything - * to take back. See {@link InsertIntoTableCommand#runInternal} for the one path that sets it and - * {@code InsertOverwriteTableCommand#run} for the caller that reads it. + * Set when the insert's transaction has committed: what it wrote -- its rows, and any table stream offset + * update it carried -- is durable. That is not the same as the insert having succeeded. A load whose + * publication times out after its commit is committed, while the session's visibility-timeout mode reports + * it to the client as an error; and a load whose plan folded to an empty relation commits nothing at all, + * since it runs no transaction unless it has an offset to commit. + * + *

Whoever owns a transaction records it where that commit happens; see + * {@code AbstractInsertExecutor#markCommitted()}. A caller that owns what happens next -- an overwrite + * publishing its temporary partitions -- decides by it: what is durable has to be published, and where + * nothing is durable a failure or a cancellation still has everything to take back. */ - private boolean committedNothing = false; + private boolean committed = false; - public boolean hasCommittedNothing() { - return committedNothing; + public boolean hasCommitted() { + return committed; } - public void setCommittedNothing(boolean committedNothing) { - this.committedNothing = committedNothing; + public void setCommitted(boolean committed) { + this.committed = committed; } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java index 065b6b03db1767..6c2d229b860776 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java @@ -756,10 +756,10 @@ private void runInternal(ConnectContext ctx, StmtExecutor executor) throws Excep AbstractInsertExecutor insertExecutor = initPlan(ctx, executor); // An empty Table Stream read still needs to commit its offset update atomically. if (!insertExecutor.requiresTransaction()) { - // Nothing was committed here and nothing will be, so a caller that treats a cancellation - // arriving after this insert as too late has nothing to keep: it has to know that the rows it - // was going to publish do not exist. See InsertCommandContext#setCommittedNothing. - insertCtx.ifPresent(insertCommandContext -> insertCommandContext.setCommittedNothing(true)); + // Nothing is committed on this path, so a caller that asks whether the rows are durable is told + // they are not -- no row and no offset was written, which is what it needs before treating a + // cancellation or a later failure here as too late to take back. + // See InsertCommandContext#setCommitted. return; } if (insertExecutorListener != null) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java index 402f3a42371188..0187f039ea77ba 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java @@ -287,7 +287,7 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { // partitions and return. for transactional, the replacement will really occur when insert successed, // i.e. `insertInto` finished. then we call taskGroupSuccess to make replacement. InsertCommandContext insertCtx = insertIntoAutoDetect(ctx, executor, taskId); - if (isCancelled.get() && insertCtx.hasCommittedNothing()) { + if (isCancelled.get() && !insertCtx.hasCommitted()) { // The load committed no row, so the cancellation still has everything to take back: the // catch drops the group's temp partitions (there are none), and the statement fails // rather than publishing a replacement the client cancelled. @@ -320,12 +320,12 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { InsertCommandContext insertCtx = insertIntoPartitions(ctx, executor, tempPartitionNames, wholeTable); cancelTheOverwriteAt(DEBUG_POINT_CANCEL_BETWEEN_THE_HALVES_OF_AN_OVERWRITE, targetTable); if (isCancelled.get()) { - if (insertCtx.hasCommittedNothing()) { - // The insert took the path that begins no transaction -- its plan folded to an empty - // relation and there was no stream offset to commit -- so this window holds no durable - // work at all. Completing the swap would publish an empty table for a statement the - // client cancelled; the catch drops the empty temp partitions instead and the - // statement fails, which is the same boundary the cancellations above sit on. + if (!insertCtx.hasCommitted()) { + // The insert committed nothing: its plan folded to an empty relation, so it took the + // path that begins no transaction, and this window holds no durable work at all. + // Completing the swap would publish an empty table for a statement the client + // cancelled; the catch drops the empty temp partitions instead and the statement + // fails, which is the same boundary the cancellations above sit on. throw cancelledBeforeTheRowsWereCommitted("after an insert that committed nothing", ctx); } // Too late to cancel: insertIntoPartitions returns only once its transaction has committed @@ -466,6 +466,16 @@ private void runInsertCommand(LogicalPlan logicalQuery, InsertCommandContext ins Optional.of(insertCtx), Optional.empty(), false, branchName); insertCommand.run(ctx, executor); if (ctx.getState().getStateType() == MysqlStateType.ERR) { + if (insertCtx.hasCommitted()) { + // The rows are durable and only their publication timed out, which the session's + // visibility-timeout mode turns into this error (`insert_visible_timeout_return_mode=error`). + // Dropping the temp partitions for it would lose exactly what the error says was committed, + // so the overwrite keeps going: the swap below publishes the rows, and the error the client + // gets stays what it is -- a statement about visibility, not about whether the overwrite ran. + LOG.info("insert overwrite continues over an error state whose rows are committed, queryId: {}", + ctx.getQueryIdentifier()); + return; + } String errMsg = Strings.emptyToNull(ctx.getState().getErrorMessage()); LOG.warn("InsertInto state error:{}", errMsg); throw new UserException(errMsg); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java index 29b80a91aad6be..70387f50619885 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java @@ -240,10 +240,14 @@ protected void onComplete() throws UserException { ctx.getSessionVariable().getInsertVisibleTimeoutMs(), txnCommitAttachment, streamUpdateInfos)) { txnStatus = TransactionStatus.VISIBLE; + markCommitted(); } else { // Keep the committed status so load accounting and insert result bookkeeping stay aligned. txnStatus = TransactionStatus.COMMITTED; publishTimedOutAfterCommit = true; + // Committed, visible later: the rows are durable even though the session's visibility-timeout + // mode may report the timeout as an error. See InsertCommandContext#setCommitted. + markCommitted(); } if (Config.isCloudMode()) { String clusterName = ctx.getCloudCluster(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/RemoteOlapInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/RemoteOlapInsertExecutor.java index c95c3b0d71d7aa..511a0a5bd4e836 100755 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/RemoteOlapInsertExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/RemoteOlapInsertExecutor.java @@ -208,6 +208,7 @@ protected void onComplete() throws UserException { } else { txnStatus = TransactionStatus.COMMITTED; } + markCommitted(); LOG.info("commit remote txn success, catalog={}, dbId={}, txnId={}, status={}", remoteCatalog.getName(), database.getId(), txnId, txnStatus); } else { diff --git a/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out index 1553c77f726d2f..4b073222d3abbe 100644 --- a/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out +++ b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out @@ -10,3 +10,8 @@ -- !dst_after_the_cancelled_empty_overwrite -- 1 2026-01-10 100 +-- !dst_after_the_publish_timeout -- +3 2026-01-20 300 +4 2026-02-20 400 +5 2026-01-25 500 + diff --git a/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy index 6361cd98f5512b..0fc4cbb33b4a2d 100644 --- a/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy +++ b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy @@ -16,8 +16,8 @@ // under the License. /** - * What a cancelled overwrite owes the rows it has already committed, and what it owes the caller when it has - * committed nothing. + * What an overwrite owes the rows it has already committed, whether it is cancelled or comes back with an + * error, and what it owes the caller when it has committed nothing. * *

An overwrite commits its rows into temporary partitions and publishes them with a swap afterwards, so a * cancellation that lands between the two halves cannot take the rows back: they are durable, and everything @@ -28,16 +28,24 @@ * cancellation still has everything to take back, and completing the swap there would publish an empty table * for a statement the client cancelled. * - *

Each case is injected by its own debug point in the overwrite, scoped by table name, so it lands on a - * real statement rather than on a race. The same window is asserted twice: once for the rows it publishes, - * once for the empty plan, whose statement fails only if the injection reached the window. + *

The same boundary decides what an error response may do to those rows: a load whose publication times out + * after its commit is committed, and the session's visibility-timeout mode reports that timeout as an error. + * The response is an error, but the rows exist, so the swap has to run for that case too. + * + *

The cancellation cases are injected by their own debug point in the overwrite, scoped by table name, so + * they land on real statements rather than on a race; the window is asserted twice, once for the rows it + * publishes and once for the empty plan, whose statement fails only if the injection reached the window. The + * publication timeout is driven by blocking the publish daemon, as test_insert_visible_timeout_return_mode + * does. */ suite("test_insert_overwrite_cancel", "nonConcurrent") { def beforePoint = "InsertOverwriteTableCommand.cancelBeforeTheInsertOfAnOverwrite" def betweenPoint = "InsertOverwriteTableCommand.cancelBetweenTheTwoHalvesOfAnOverwrite" + def stopPublishPoint = "PublishVersionDaemon.stop_publish" GetDebugPoint().disableDebugPointForAllFEs(beforePoint) GetDebugPoint().disableDebugPointForAllFEs(betweenPoint) + GetDebugPoint().disableDebugPointForAllFEs(stopPublishPoint) sql """DROP TABLE IF EXISTS test_iot_cancel_src""" sql """DROP TABLE IF EXISTS test_iot_cancel_dst""" sql """DROP TABLE IF EXISTS test_iot_cancel_flat_src""" @@ -155,4 +163,37 @@ suite("test_insert_overwrite_cancel", "nonConcurrent") { GetDebugPoint().disableDebugPointForAllFEs(betweenPoint) } order_qt_dst_after_the_cancelled_empty_overwrite """SELECT id, dt, amount FROM test_iot_cancel_flat_dst""" + + // An error response over committed rows: `insert_visible_timeout_return_mode=error` turns a publication + // timeout that follows the commit into an error, but the rows the overwrite wrote are durable, so the + // overwrite has to be published rather than dropped with the temporary partitions. The row the overwrite + // reads is new, so the table only holds it if the swap ran. + sql """INSERT INTO test_iot_cancel_src VALUES (5, '2026-01-25', 500)""" + try { + GetDebugPoint().enableDebugPointForAllFEs(stopPublishPoint, [timeout: "10"]) + sql """SET insert_visible_timeout_ms = 1000""" + sql """SET insert_visible_timeout_return_mode = 'error'""" + test { + sql """INSERT OVERWRITE TABLE test_iot_cancel_dst SELECT * FROM test_iot_cancel_src""" + exception "transaction commit successfully, BUT data did not become visible" + } + } finally { + GetDebugPoint().disableDebugPointForAllFEs(stopPublishPoint) + // Back to the defaults (DEFAULT_INSERT_VISIBLE_TIMEOUT_MS = 60000, 'committed'). + sql """SET insert_visible_timeout_ms = 60000""" + sql """SET insert_visible_timeout_return_mode = 'committed'""" + } + + // Publish resumes on its own, and the rows that were committed under the error come with it. + def published = false + for (int i = 0; i < 15; i++) { + def rows = sql """SELECT count(*) FROM test_iot_cancel_dst""" + if ((rows[0][0] as long) == 3L) { + published = true + break + } + sleep(1000) + } + assertTrue(published, "The committed overwrite should become visible after publish resumes") + order_qt_dst_after_the_publish_timeout """SELECT id, dt, amount FROM test_iot_cancel_dst""" } From 4b43d5ba6a633c879d9531b79209da50c4f8be86 Mon Sep 17 00:00:00 2001 From: yujun Date: Wed, 30 Sep 2026 15:24:20 +0800 Subject: [PATCH 4/9] [fix](fe) Honour a cancellation that reaches the swap while it waits for the lock The last read of the cancellation flag happens before the swap is issued, and the swap then waits for the target table's write lock -- a wait that lasts as long as whoever holds it. A cancellation that found nothing committed could therefore still be followed by a swap publishing empty temporary partitions over the rows the client asked to keep. Key changes: - The flag is read once more under that lock, where the wait happens, and the swap is refused when a cancellation found nothing committed -- there is nothing durable to publish, so refusing costs the statement and leaves the rows. Only a local table is wrapped: a remote table swaps on the frontend that owns it. - A debug point injects the cancellation there, and the suite asserts the table keeps its rows. - The publication-timeout case is skipped in cloud mode, where the commit has no FE publication wait to time out, as test_insert_visible_timeout_return_mode does. The cancellation cases stay active there. Unit Test: - regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy - insert_overwrite_p0 (14 suites), insert_p0/test_insert_visible_timeout_return_mode and mtmv_p0/ivm/test_ivm_overwrite_failure_between_the_halves pass. --- .../insert/InsertOverwriteTableCommand.java | 49 ++++++++++- .../test_insert_overwrite_cancel.out | 3 + .../test_insert_overwrite_cancel.groovy | 84 ++++++++++++------- 3 files changed, 106 insertions(+), 30 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java index 0187f039ea77ba..b8bad13acf3c48 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java @@ -136,6 +136,16 @@ public class InsertOverwriteTableCommand extends Command public static final String DEBUG_POINT_CANCEL_BETWEEN_THE_HALVES_OF_AN_OVERWRITE = "InsertOverwriteTableCommand.cancelBetweenTheTwoHalvesOfAnOverwrite"; + /** + * Cancels the overwrite while the swap holds the target table's write lock, which is the window a + * cancellation can reach only after the check that reads the flag before the swap was issued: the swap + * waits for that lock, and the wait can be as long as whoever holds it. A cancellation with nothing + * committed is still honoured there, because there is nothing durable to publish and refusing costs the + * statement and nothing else. See test_insert_overwrite_cancel. + */ + public static final String DEBUG_POINT_CANCEL_WHILE_THE_SWAP_WAITS_FOR_THE_TABLE_LOCK = + "InsertOverwriteTableCommand.cancelWhileTheSwapWaitsForTheTableLock"; + private static final Logger LOG = LogManager.getLogger(InsertOverwriteTableCommand.class); private LogicalPlan originLogicalQuery; @@ -338,8 +348,7 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { + " queryId: {}", ctx.getQueryIdentifier()); } failBetweenTheTwoHalvesOfAnOverwrite(targetTable); - InsertOverwriteUtil.replacePartition(targetTable, partitionNames, tempPartitionNames, - isForceDropPartition()); + publishTheOverwrite(targetTable, partitionNames, tempPartitionNames, insertCtx, ctx); if (isCancelled.get()) { LOG.info("insert overwrite is cancelled before taskSuccess, do nothing, queryId: {}", ctx.getQueryIdentifier()); @@ -449,6 +458,42 @@ private void cancelTheOverwriteAt(String debugPointName, TableIf targetTable) { cancel(); } + /** + * Publishes this overwrite by swapping the temp partitions in, with a last look at the cancellation flag + * taken under the lock the swap contends for. + * + *

{@link #run} reads the flag before the swap is issued, and the swap then waits for the table's write + * lock, so a cancellation that arrives during that wait is the one place a check before the swap cannot + * see. Reading it again here costs nothing and is where the wait happens: for a cancellation with nothing + * committed there is nothing durable to publish, so refusing to swap costs the statement and leaves the + * rows the client asked to keep -- while a swap that went ahead would replace them with empty partitions. + * + *

Only a local table is wrapped: a remote table swaps on the frontend that owns it, where this lock + * says nothing. + */ + private void publishTheOverwrite(TableIf targetTable, List partitionNames, + List tempPartitionNames, InsertCommandContext insertCtx, ConnectContext ctx) throws UserException { + if (!(targetTable instanceof OlapTable) || targetTable instanceof RemoteOlapTable) { + InsertOverwriteUtil.replacePartition(targetTable, partitionNames, tempPartitionNames, + isForceDropPartition()); + return; + } + OlapTable olapTable = (OlapTable) targetTable; + if (!olapTable.writeLockIfExist()) { + return; + } + try { + cancelTheOverwriteAt(DEBUG_POINT_CANCEL_WHILE_THE_SWAP_WAITS_FOR_THE_TABLE_LOCK, targetTable); + if (isCancelled.get() && !insertCtx.hasCommitted()) { + throw cancelledBeforeTheRowsWereCommitted("while the swap waited for the table lock", ctx); + } + InsertOverwriteUtil.replacePartition(targetTable, partitionNames, tempPartitionNames, + isForceDropPartition()); + } finally { + olapTable.writeUnlock(); + } + } + /** * The failure a cancellation that found nothing durable is reported as. No row and no stream offset was * committed, so a re-run reads the same rows -- which is why this is a failure rather than the success of diff --git a/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out index 4b073222d3abbe..a4d590529277ee 100644 --- a/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out +++ b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out @@ -10,6 +10,9 @@ -- !dst_after_the_cancelled_empty_overwrite -- 1 2026-01-10 100 +-- !dst_after_the_cancelled_locked_swap -- +1 2026-01-10 100 + -- !dst_after_the_publish_timeout -- 3 2026-01-20 300 4 2026-02-20 400 diff --git a/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy index 0fc4cbb33b4a2d..e34fe7c6d109f8 100644 --- a/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy +++ b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy @@ -32,19 +32,25 @@ * after its commit is committed, and the session's visibility-timeout mode reports that timeout as an error. * The response is an error, but the rows exist, so the swap has to run for that case too. * + *

Where a cancellation lands inside the window, but the swap is already waiting for the table's write lock, + * the flag is read once more under that lock: nothing is durable, so the overwrite can still be refused there + * instead of publishing empty partitions. + * *

The cancellation cases are injected by their own debug point in the overwrite, scoped by table name, so - * they land on real statements rather than on a race; the window is asserted twice, once for the rows it - * publishes and once for the empty plan, whose statement fails only if the injection reached the window. The - * publication timeout is driven by blocking the publish daemon, as test_insert_visible_timeout_return_mode - * does. + * they land on real statements rather than on a race; the window is asserted for the rows it publishes, for + * the empty plan whose statement fails only if the injection reached the window, and for the lock the swap + * waits on. The publication timeout is driven by blocking the publish daemon, as + * test_insert_visible_timeout_return_mode does. */ suite("test_insert_overwrite_cancel", "nonConcurrent") { def beforePoint = "InsertOverwriteTableCommand.cancelBeforeTheInsertOfAnOverwrite" def betweenPoint = "InsertOverwriteTableCommand.cancelBetweenTheTwoHalvesOfAnOverwrite" + def waitsPoint = "InsertOverwriteTableCommand.cancelWhileTheSwapWaitsForTheTableLock" def stopPublishPoint = "PublishVersionDaemon.stop_publish" GetDebugPoint().disableDebugPointForAllFEs(beforePoint) GetDebugPoint().disableDebugPointForAllFEs(betweenPoint) + GetDebugPoint().disableDebugPointForAllFEs(waitsPoint) GetDebugPoint().disableDebugPointForAllFEs(stopPublishPoint) sql """DROP TABLE IF EXISTS test_iot_cancel_src""" sql """DROP TABLE IF EXISTS test_iot_cancel_dst""" @@ -164,36 +170,58 @@ suite("test_insert_overwrite_cancel", "nonConcurrent") { } order_qt_dst_after_the_cancelled_empty_overwrite """SELECT id, dt, amount FROM test_iot_cancel_flat_dst""" - // An error response over committed rows: `insert_visible_timeout_return_mode=error` turns a publication - // timeout that follows the commit into an error, but the rows the overwrite wrote are durable, so the - // overwrite has to be published rather than dropped with the temporary partitions. The row the overwrite - // reads is new, so the table only holds it if the swap ran. - sql """INSERT INTO test_iot_cancel_src VALUES (5, '2026-01-25', 500)""" + // The last chance to honour a cancellation that found nothing committed: this one is delivered while the + // swap holds the table's write lock, after the check that reads the flag before the swap was issued. The + // table has to keep its rows, because a swap there would publish the empty partitions. try { - GetDebugPoint().enableDebugPointForAllFEs(stopPublishPoint, [timeout: "10"]) - sql """SET insert_visible_timeout_ms = 1000""" - sql """SET insert_visible_timeout_return_mode = 'error'""" + GetDebugPoint().enableDebugPointForAllFEs(waitsPoint, + [table_name: "test_iot_cancel_flat_dst"]) test { - sql """INSERT OVERWRITE TABLE test_iot_cancel_dst SELECT * FROM test_iot_cancel_src""" - exception "transaction commit successfully, BUT data did not become visible" + sql """INSERT OVERWRITE TABLE test_iot_cancel_flat_dst + SELECT id, dt, amount FROM test_iot_cancel_flat_src WHERE 1 = 0""" + exception "insert overwrite is cancelled while the swap waited for the table lock" } } finally { - GetDebugPoint().disableDebugPointForAllFEs(stopPublishPoint) - // Back to the defaults (DEFAULT_INSERT_VISIBLE_TIMEOUT_MS = 60000, 'committed'). - sql """SET insert_visible_timeout_ms = 60000""" - sql """SET insert_visible_timeout_return_mode = 'committed'""" + GetDebugPoint().disableDebugPointForAllFEs(waitsPoint) } + order_qt_dst_after_the_cancelled_locked_swap """SELECT id, dt, amount FROM test_iot_cancel_flat_dst""" + + // An error response over committed rows: `insert_visible_timeout_return_mode=error` turns a publication + // timeout that follows the commit into an error, but the rows the overwrite wrote are durable, so the + // overwrite has to be published rather than dropped with the temporary partitions. The row the overwrite + // reads is new, so the table only holds it if the swap ran. + // + // Cloud mode is skipped: its commit has no FE publication wait (`CloudGlobalTransactionMgr` commits + // straight into the meta service), so blocking the publish daemon leaves the insert reporting success and + // there is no timeout to reproduce. The cancellation cases above do not depend on that and stay active. + if (!isCloudMode()) { + sql """INSERT INTO test_iot_cancel_src VALUES (5, '2026-01-25', 500)""" + try { + GetDebugPoint().enableDebugPointForAllFEs(stopPublishPoint, [timeout: "10"]) + sql """SET insert_visible_timeout_ms = 1000""" + sql """SET insert_visible_timeout_return_mode = 'error'""" + test { + sql """INSERT OVERWRITE TABLE test_iot_cancel_dst SELECT * FROM test_iot_cancel_src""" + exception "transaction commit successfully, BUT data did not become visible" + } + } finally { + GetDebugPoint().disableDebugPointForAllFEs(stopPublishPoint) + // Back to the defaults (DEFAULT_INSERT_VISIBLE_TIMEOUT_MS = 60000, 'committed'). + sql """SET insert_visible_timeout_ms = 60000""" + sql """SET insert_visible_timeout_return_mode = 'committed'""" + } - // Publish resumes on its own, and the rows that were committed under the error come with it. - def published = false - for (int i = 0; i < 15; i++) { - def rows = sql """SELECT count(*) FROM test_iot_cancel_dst""" - if ((rows[0][0] as long) == 3L) { - published = true - break + // Publish resumes on its own, and the rows that were committed under the error come with it. + def published = false + for (int i = 0; i < 15; i++) { + def rows = sql """SELECT count(*) FROM test_iot_cancel_dst""" + if ((rows[0][0] as long) == 3L) { + published = true + break + } + sleep(1000) } - sleep(1000) + assertTrue(published, "The committed overwrite should become visible after publish resumes") + order_qt_dst_after_the_publish_timeout """SELECT id, dt, amount FROM test_iot_cancel_dst""" } - assertTrue(published, "The committed overwrite should become visible after publish resumes") - order_qt_dst_after_the_publish_timeout """SELECT id, dt, amount FROM test_iot_cancel_dst""" } From 5eee943f3d313befe4c815a58178ea4f0e7610b6 Mon Sep 17 00:00:00 2001 From: yujun Date: Wed, 30 Sep 2026 15:31:45 +0800 Subject: [PATCH 5/9] [fix](fe) Correct the window comment for a publication that timed out The comment on the window's "too late to cancel" branch still described the insert as having published the rows by the time it returns. Since the publication-timeout case is handled there too, the rows can be committed with their publication still pending, which is what the comment now says. --- .../commands/insert/InsertOverwriteTableCommand.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java index b8bad13acf3c48..b213a45f0a5701 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java @@ -339,11 +339,11 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { throw cancelledBeforeTheRowsWereCommitted("after an insert that committed nothing", ctx); } // Too late to cancel: insertIntoPartitions returns only once its transaction has committed - // and published the rows into the temp partitions, and everything the read consumed -- the - // base table stream offsets among it -- was committed with that same transaction. Dropping - // the temp partitions here is exactly what would lose those rows against an advanced - // offset, while the swap below is what publishes them. The overwrite completes, and it is - // the outcome the statement reports. + // the rows into the temp partitions -- visible, or still waiting for a publication that + // timed out -- and everything the read consumed, the base table stream offsets among it, + // was committed with that same transaction. Dropping the temp partitions here is exactly + // what would lose those rows against an advanced offset, while the swap below is what + // publishes them. The overwrite completes, and it is the outcome the statement reports. LOG.info("insert overwrite is cancelled after its rows were committed, completing it," + " queryId: {}", ctx.getQueryIdentifier()); } From b2e70317465b16e01534b8fb32bfedc4a8dd24dc Mon Sep 17 00:00:00 2001 From: yujun Date: Wed, 30 Sep 2026 16:16:28 +0800 Subject: [PATCH 6/9] [fix](fe) Fail an overwrite whose target was dropped before the swap The last-chance wrapper took the target table's write lock itself and returned silently when the table no longer existed, after which run() acknowledged the overwrite with a success. Before the wrapper, the utility's own early return for a dropped table ran into a finally that unlocks a lock that return never took, so it raised and the statement failed. That outcome is the one a client whose swap never happened is owed, and it is now taken deliberately, with a message naming the table. Key changes: - publishTheOverwrite raises for a dropped target instead of skipping the swap. The existing catch drops the temp partitions of the dropped table and takes the task back. Unit Test: - regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy. The dropped-target path needs a concurrent DROP and has no injection point of its own, so it is reasoned rather than driven. - test_insert_overwrite_cancel, test_iot_range_nereids, test_iot_auto_detect and mtmv_p0/ivm/test_ivm_basic_mtmv pass. --- .../commands/insert/InsertOverwriteTableCommand.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java index b213a45f0a5701..4d766e9af06598 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java @@ -480,7 +480,13 @@ private void publishTheOverwrite(TableIf targetTable, List partitionName } OlapTable olapTable = (OlapTable) targetTable; if (!olapTable.writeLockIfExist()) { - return; + // The target was dropped while this overwrite ran, so there is nothing to publish into and no swap + // to issue. Failing is also what the utility's own early return did for a dropped table -- its + // finally unlocks a lock that return never took, which raises -- and it is what a client whose + // swap never happened is owed: acknowledging the overwrite would claim rows the table cannot hold. + // The catch drops the temp partitions of the dropped table and takes the task back. + throw new UserException("insert overwrite could not publish its temporary partitions: table " + + olapTable.getName() + " was dropped, queryId: " + ctx.getQueryIdentifier()); } try { cancelTheOverwriteAt(DEBUG_POINT_CANCEL_WHILE_THE_SWAP_WAITS_FOR_THE_TABLE_LOCK, targetTable); From 9039a08b9caaa367c748d43cf682bdfb9f9d8e3c Mon Sep 17 00:00:00 2001 From: yujun Date: Wed, 30 Sep 2026 16:28:27 +0800 Subject: [PATCH 7/9] [test](fe) Pin the overwrite swap's failure for a dropped target publishTheOverwrite raises when it cannot take the target table's write lock, which is the state a concurrent DROP leaves. The branch is driven directly, because reaching it through a statement needs a DROP concurrent with the swap and the command has no injection point for that; the condition the branch reads is what the mock supplies. Unit Test: - InsertOverwriteDroppedTargetTest#targetDroppedBeforeTheSwapFailsTheOverwrite (Tests run: 1, Failures: 0) --- .../InsertOverwriteDroppedTargetTest.java | 69 +++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteDroppedTargetTest.java diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteDroppedTargetTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteDroppedTargetTest.java new file mode 100644 index 00000000000000..8464f60031bf00 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteDroppedTargetTest.java @@ -0,0 +1,69 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.trees.plans.commands.insert; + +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.common.UserException; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; +import org.apache.doris.qe.ConnectContext; + +import com.google.common.collect.Lists; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Optional; + +/** + * What the swap of an overwrite does when its target has been dropped underneath it. + * + *

The swap takes the target table's write lock, and a target dropped by a concurrent DROP no longer + * offers one. Skipping the swap there and returning lets {@code run} acknowledge the overwrite with a + * success for a table that holds nothing it wrote, which is the one outcome a client must not be told; + * {@link InsertOverwriteTableCommand} raises instead, so the caller's catch drops the temp partitions and + * the statement fails. This pins that branch: the table a lock could not be taken on has to be named in + * the failure. + * + *

Driven directly rather than through a statement because the window needs a DROP concurrent with the + * swap, which a regression test cannot place without an injection point of its own; the condition the + * branch reads -- {@code writeLockIfExist()} answering false -- is what the mock supplies. + */ +public class InsertOverwriteDroppedTargetTest { + + @Test + public void targetDroppedBeforeTheSwapFailsTheOverwrite() { + InsertOverwriteTableCommand command = new InsertOverwriteTableCommand( + Mockito.mock(LogicalPlan.class), Optional.empty(), Optional.empty(), Optional.empty()); + OlapTable droppedTarget = Mockito.mock(OlapTable.class); + Mockito.when(droppedTarget.getName()).thenReturn("iot_dropped_target"); + Mockito.when(droppedTarget.writeLockIfExist()).thenReturn(false); + ConnectContext ctx = Mockito.mock(ConnectContext.class); + Mockito.when(ctx.getQueryIdentifier()).thenReturn("stmt[1, query-id]"); + + UserException thrown = Assertions.assertThrows(UserException.class, + () -> Deencapsulation.invoke(command, "publishTheOverwrite", droppedTarget, + Lists.newArrayList("p1"), Lists.newArrayList("tp1"), + new OlapInsertCommandContext(false, true), ctx)); + Assertions.assertTrue(thrown.getMessage() != null + && thrown.getMessage().contains("iot_dropped_target") + && thrown.getMessage().contains("was dropped"), + "the failure has to name the table whose swap could not be issued, but was: " + + thrown.getMessage()); + } +} From b2f346cee90786f58cd513d0afaea8cd852eaf31 Mon Sep 17 00:00:00 2001 From: yujun Date: Wed, 30 Sep 2026 17:08:48 +0800 Subject: [PATCH 8/9] [fix](fe) Take the last-chance cancellation check on the auto-detect route too An auto-detect overwrite (PARTITION(*)) publishes through the overwrite manager, so it bypassed the check the explicit-partition route got. With enable_strict_consistency_dml off, an empty-plan auto-detect overwrite has no exchange in its plan -- VEMPTYSET is the sink's direct child -- so it takes the insert's no-transaction path, and a cancellation landing while the swap waits for the table lock was answered with a success for a statement that did nothing. Both routes now publish through one place that holds the target table's write lock and reads the flag under it. Key changes: - publishTheOverwrite takes the publication as an action, so the explicit-partition swap and the auto-detect task group share the same last-chance decision, and the same failure for a target that was dropped. - The suite drives the auto-detect shape: with strict DML consistency off, a cancelled PARTITION(*) overwrite of an empty plan fails and the table keeps its rows. Unit Test: - regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy - test_insert_overwrite_cancel, test_iot_auto_detect, test_iot_auto_detect_fail, test_iot_auto_detect_concurrent, test_iot_overwrite_and_create and mtmv_p0/ivm/test_ivm_basic_mtmv pass. --- .../insert/InsertOverwriteTableCommand.java | 51 +++++++++++++------ .../test_insert_overwrite_cancel.out | 4 ++ .../test_insert_overwrite_cancel.groovy | 19 +++++++ 3 files changed, 58 insertions(+), 16 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java index 4d766e9af06598..36d246128f012d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java @@ -146,6 +146,17 @@ public class InsertOverwriteTableCommand extends Command public static final String DEBUG_POINT_CANCEL_WHILE_THE_SWAP_WAITS_FOR_THE_TABLE_LOCK = "InsertOverwriteTableCommand.cancelWhileTheSwapWaitsForTheTableLock"; + /** + * The swap that publishes an overwrite: replacing the temp partitions for an explicit-partition + * overwrite, or making a task group's replacements visible for an auto-detect one. Both run through + * {@link #publishTheOverwrite}, which holds the target table's write lock and takes the last look at the + * cancellation flag before letting one run. + */ + @FunctionalInterface + private interface OverwritePublication { + void publish() throws UserException; + } + private static final Logger LOG = LogManager.getLogger(InsertOverwriteTableCommand.class); private LogicalPlan originLogicalQuery; @@ -292,18 +303,23 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { } if (isAutoDetectOverwrite(getLogicalQuery())) { // taskId here is a group id. it contains all replace tasks made and registered in rpc process. - taskId = insertOverwriteManager.registerTaskGroup(targetTable); + final long groupId = insertOverwriteManager.registerTaskGroup(targetTable); + taskId = groupId; // When inserting, BE will call to replace partition by FrontendService. FE will register new temp // partitions and return. for transactional, the replacement will really occur when insert successed, // i.e. `insertInto` finished. then we call taskGroupSuccess to make replacement. - InsertCommandContext insertCtx = insertIntoAutoDetect(ctx, executor, taskId); + InsertCommandContext insertCtx = insertIntoAutoDetect(ctx, executor, groupId); if (isCancelled.get() && !insertCtx.hasCommitted()) { - // The load committed no row, so the cancellation still has everything to take back: the - // catch drops the group's temp partitions (there are none), and the statement fails - // rather than publishing a replacement the client cancelled. + // The load committed nothing -- an empty plan takes the path that begins no transaction -- + // so the cancellation still has everything to take back: the catch drops the group's temp + // partitions (an empty plan registers none), and the statement fails rather than reporting + // a replacement the client cancelled. A cancellation landing after this check, while the + // swap waits for the table lock, is taken up again by publishTheOverwrite below. throw cancelledBeforeTheRowsWereCommitted("after a load that committed nothing", ctx); } - insertOverwriteManager.taskGroupSuccess(taskId, (OlapTable) targetTable, isForceDropPartition()); + publishTheOverwrite(targetTable, insertCtx, ctx, + () -> insertOverwriteManager.taskGroupSuccess(groupId, (OlapTable) targetTable, + isForceDropPartition())); } else { // it's overwrite table(as all partitions) or specific partition(s) List tempPartitionNames = InsertOverwriteUtil.generateTempPartitionNames(partitionNames); @@ -348,7 +364,12 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { + " queryId: {}", ctx.getQueryIdentifier()); } failBetweenTheTwoHalvesOfAnOverwrite(targetTable); - publishTheOverwrite(targetTable, partitionNames, tempPartitionNames, insertCtx, ctx); + // The publication below is a lambda, so the partitions it replaces need a name that is final: + // partitionNames is assigned on more than one path above. + final List replacedPartitionNames = partitionNames; + publishTheOverwrite(targetTable, insertCtx, ctx, + () -> InsertOverwriteUtil.replacePartition(targetTable, replacedPartitionNames, + tempPartitionNames, isForceDropPartition())); if (isCancelled.get()) { LOG.info("insert overwrite is cancelled before taskSuccess, do nothing, queryId: {}", ctx.getQueryIdentifier()); @@ -459,23 +480,22 @@ private void cancelTheOverwriteAt(String debugPointName, TableIf targetTable) { } /** - * Publishes this overwrite by swapping the temp partitions in, with a last look at the cancellation flag - * taken under the lock the swap contends for. + * Publishes this overwrite by running its swap, with a last look at the cancellation flag taken under the + * lock the swap contends for. * *

{@link #run} reads the flag before the swap is issued, and the swap then waits for the table's write * lock, so a cancellation that arrives during that wait is the one place a check before the swap cannot * see. Reading it again here costs nothing and is where the wait happens: for a cancellation with nothing * committed there is nothing durable to publish, so refusing to swap costs the statement and leaves the - * rows the client asked to keep -- while a swap that went ahead would replace them with empty partitions. + * rows the client asked to keep -- while a swap that went ahead would replace them with an empty result. * *

Only a local table is wrapped: a remote table swaps on the frontend that owns it, where this lock * says nothing. */ - private void publishTheOverwrite(TableIf targetTable, List partitionNames, - List tempPartitionNames, InsertCommandContext insertCtx, ConnectContext ctx) throws UserException { + private void publishTheOverwrite(TableIf targetTable, InsertCommandContext insertCtx, ConnectContext ctx, + OverwritePublication publication) throws UserException { if (!(targetTable instanceof OlapTable) || targetTable instanceof RemoteOlapTable) { - InsertOverwriteUtil.replacePartition(targetTable, partitionNames, tempPartitionNames, - isForceDropPartition()); + publication.publish(); return; } OlapTable olapTable = (OlapTable) targetTable; @@ -493,8 +513,7 @@ private void publishTheOverwrite(TableIf targetTable, List partitionName if (isCancelled.get() && !insertCtx.hasCommitted()) { throw cancelledBeforeTheRowsWereCommitted("while the swap waited for the table lock", ctx); } - InsertOverwriteUtil.replacePartition(targetTable, partitionNames, tempPartitionNames, - isForceDropPartition()); + publication.publish(); } finally { olapTable.writeUnlock(); } diff --git a/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out index a4d590529277ee..90d04658c0abea 100644 --- a/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out +++ b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out @@ -13,6 +13,10 @@ -- !dst_after_the_cancelled_locked_swap -- 1 2026-01-10 100 +-- !dst_after_the_cancelled_auto_detect_swap -- +3 2026-01-20 300 +4 2026-02-20 400 + -- !dst_after_the_publish_timeout -- 3 2026-01-20 300 4 2026-02-20 400 diff --git a/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy index e34fe7c6d109f8..f83314073dd629 100644 --- a/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy +++ b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy @@ -186,6 +186,25 @@ suite("test_insert_overwrite_cancel", "nonConcurrent") { } order_qt_dst_after_the_cancelled_locked_swap """SELECT id, dt, amount FROM test_iot_cancel_flat_dst""" + // The auto-detect route takes the same decision at the same point, and it needs it: with + // enable_strict_consistency_dml off, an empty-plan `PARTITION(*)` overwrite has no exchange in its plan + // either, so it takes the insert's no-transaction path as well and its swap has nothing to publish. The + // cancellation has to be honoured there too, and the table has to keep its rows. + try { + sql """SET enable_strict_consistency_dml = false""" + GetDebugPoint().enableDebugPointForAllFEs(waitsPoint, + [table_name: "test_iot_cancel_dst"]) + test { + sql """INSERT OVERWRITE TABLE test_iot_cancel_dst PARTITION(*) + SELECT id, dt, amount FROM test_iot_cancel_src WHERE 1 = 0""" + exception "insert overwrite is cancelled while the swap waited for the table lock" + } + } finally { + GetDebugPoint().disableDebugPointForAllFEs(waitsPoint) + sql """SET enable_strict_consistency_dml = true""" + } + order_qt_dst_after_the_cancelled_auto_detect_swap """SELECT id, dt, amount FROM test_iot_cancel_dst""" + // An error response over committed rows: `insert_visible_timeout_return_mode=error` turns a publication // timeout that follows the commit into an error, but the rows the overwrite wrote are durable, so the // overwrite has to be published rather than dropped with the temporary partitions. The row the overwrite From 1c287182321e75780125165061900374df220b7d Mon Sep 17 00:00:00 2001 From: yujun Date: Wed, 30 Sep 2026 17:41:50 +0800 Subject: [PATCH 9/9] [fix](fe) Split the auto-detect group's bookkeeping out of the table lock Two follow-ups from the previous commit. The task-group publication ran all of its bookkeeping under the target table's write lock, which the wrapper holds: that bookkeeping writes an edit-log entry for every task (an ADD and a DROP each), and with the default journal durability (SYNC/SYNC/SIMPLE_MAJORITY) every one of them waits for a replica acknowledgement, so a PARTITION(*) overwrite spanning many partitions kept readers and writers of the target blocked through all of those waits. And the unit test added for the dropped-target branch still called the publication method with its pre-refactor signature, so its reflective call never reached the branch it was written for. Key changes: - taskGroupSuccess is split into the replacement and the bookkeeping. The overwrite calls the replacement under its last-chance lock and the bookkeeping after it; the owner-side entry point keeps calling the composed method, so that path is unchanged. - The dropped-target test calls the current method with a publication callback and asserts the callback was not reached. Unit Test: - InsertOverwriteDroppedTargetTest#targetDroppedBeforeTheSwapFailsTheOverwrite (Tests run: 1, Failures: 0, Errors: 0) - test_insert_overwrite_cancel, test_iot_auto_detect, test_iot_auto_detect_fail, test_iot_auto_detect_concurrent and test_iot_overwrite_and_create pass. --- .../AbstractInsertOverwriteManager.java | 13 ++++++++++ .../InsertOverwriteManager.java | 25 +++++++++++++++++++ .../RemoteInsertOverwriteManager.java | 13 ++++++++++ .../insert/InsertOverwriteTableCommand.java | 8 ++++-- .../InsertOverwriteDroppedTargetTest.java | 12 ++++++--- 5 files changed, 65 insertions(+), 6 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/AbstractInsertOverwriteManager.java b/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/AbstractInsertOverwriteManager.java index 50321927fc5834..04088d48fb6d1e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/AbstractInsertOverwriteManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/AbstractInsertOverwriteManager.java @@ -33,6 +33,19 @@ public interface AbstractInsertOverwriteManager { void taskGroupSuccess(long groupId, OlapTable targetTable, boolean forceDropPartition) throws DdlException; + /** + * The replacement half of {@link #taskGroupSuccess}: makes every replacement the group recorded visible. + * A caller that publishes the group itself calls this and then {@link #finishTaskGroup}, so that the task + * bookkeeping -- an edit-log write per task, each waiting for its journal -- runs outside whatever lock the + * caller holds the replacement under. A manager whose owning frontend runs the whole sequence behind one + * call replaces there and finishes there. + */ + void replacePartitionsOfTaskGroup(long groupId, OlapTable targetTable, boolean forceDropPartition) + throws DdlException; + + /** The bookkeeping half of {@link #taskGroupSuccess}, after the replacement it follows. */ + void finishTaskGroup(long groupId); + void taskSuccess(long taskId) throws Exception; void taskGroupFail(long groupId) throws Exception; diff --git a/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/InsertOverwriteManager.java b/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/InsertOverwriteManager.java index 3f0f7feff41c84..9863a6fdfcadb5 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/InsertOverwriteManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/InsertOverwriteManager.java @@ -191,6 +191,21 @@ public void taskGroupFail(long groupId) { // here we will make all raplacement of this group visiable. if someone fails, nothing happen. @Override public void taskGroupSuccess(long groupId, OlapTable targetTable, boolean forceDropPartition) throws DdlException { + replacePartitionsOfTaskGroup(groupId, targetTable, forceDropPartition); + finishTaskGroup(groupId); + } + + /** + * The replacement half of {@link #taskGroupSuccess}: makes every replacement this group recorded visible. + * + *

Separate from the bookkeeping that follows it because the caller holds the target table's write lock + * for the last cancellation check before it publishes: the bookkeeping writes an edit-log entry per task, + * each of which waits for its journal to be acknowledged, and readers and writers of the target must not + * stay blocked through those waits -- the utility's own replacement, which the lock is about, does not. + */ + @Override + public void replacePartitionsOfTaskGroup(long groupId, OlapTable targetTable, boolean forceDropPartition) + throws DdlException { try { Map relations = partitionPairs.get(groupId); ArrayList oldNames = new ArrayList<>(); @@ -205,6 +220,16 @@ public void taskGroupSuccess(long groupId, OlapTable targetTable, boolean forceD + "all new partition will not be visible and will be recycled by partition GC."); throw e; } + } + + /** + * The bookkeeping half of {@link #taskGroupSuccess}: makes the group's tasks final and forgets the group, + * after the replacement has been made. See + * {@link #replacePartitionsOfTaskGroup} for why it is not the replacement's caller to hold this under a + * table lock. + */ + @Override + public void finishTaskGroup(long groupId) { LOG.info("insert overwrite auto detect partition task group [" + groupId + "] succeed"); for (Long taskId : taskGroups.get(groupId)) { Env.getCurrentEnv().getEditLog() diff --git a/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/RemoteInsertOverwriteManager.java b/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/RemoteInsertOverwriteManager.java index 4e21501c1ee143..65e3f9977fd99d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/RemoteInsertOverwriteManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/insertoverwrite/RemoteInsertOverwriteManager.java @@ -63,6 +63,19 @@ public void taskGroupSuccess(long groupId, OlapTable targetTable, boolean forceD targetTable.getDatabase().getFullName(), targetTable.getName(), groupId, forceDropPartition); } + @Override + public void replacePartitionsOfTaskGroup(long groupId, OlapTable targetTable, boolean forceDropPartition) + throws DdlException { + // The owning frontend runs the replacement and the bookkeeping behind this one call, so the caller + // does not hold its lock across them; finishTaskGroup has nothing left to do here. + taskGroupSuccess(groupId, targetTable, forceDropPartition); + } + + @Override + public void finishTaskGroup(long groupId) { + // Finished on the owning frontend, which the replacement call reached. + } + @Override public void taskSuccess(long taskId) throws Exception { catalog.getFeServiceClient().taskSuccess(taskId); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java index 36d246128f012d..01bdd3b7b2a80b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java @@ -153,7 +153,7 @@ public class InsertOverwriteTableCommand extends Command * cancellation flag before letting one run. */ @FunctionalInterface - private interface OverwritePublication { + interface OverwritePublication { void publish() throws UserException; } @@ -317,9 +317,13 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { // swap waits for the table lock, is taken up again by publishTheOverwrite below. throw cancelledBeforeTheRowsWereCommitted("after a load that committed nothing", ctx); } + // The replacement stays under the lock publishTheOverwrite takes, and the group's bookkeeping + // follows it: its edit-log writes wait for their journals, and the target's readers and + // writers must not be blocked through them. publishTheOverwrite(targetTable, insertCtx, ctx, - () -> insertOverwriteManager.taskGroupSuccess(groupId, (OlapTable) targetTable, + () -> insertOverwriteManager.replacePartitionsOfTaskGroup(groupId, (OlapTable) targetTable, isForceDropPartition())); + insertOverwriteManager.finishTaskGroup(groupId); } else { // it's overwrite table(as all partitions) or specific partition(s) List tempPartitionNames = InsertOverwriteUtil.generateTempPartitionNames(partitionNames); diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteDroppedTargetTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteDroppedTargetTest.java index 8464f60031bf00..ac1dd5e54c2ba6 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteDroppedTargetTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteDroppedTargetTest.java @@ -23,12 +23,12 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; import org.apache.doris.qe.ConnectContext; -import com.google.common.collect.Lists; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.Mockito; import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; /** * What the swap of an overwrite does when its target has been dropped underneath it. @@ -42,7 +42,8 @@ * *

Driven directly rather than through a statement because the window needs a DROP concurrent with the * swap, which a regression test cannot place without an injection point of its own; the condition the - * branch reads -- {@code writeLockIfExist()} answering false -- is what the mock supplies. + * branch reads -- {@code writeLockIfExist()} answering false -- is what the mock supplies, and the + * publication it is handed records whether it was reached at all. */ public class InsertOverwriteDroppedTargetTest { @@ -55,15 +56,18 @@ public void targetDroppedBeforeTheSwapFailsTheOverwrite() { Mockito.when(droppedTarget.writeLockIfExist()).thenReturn(false); ConnectContext ctx = Mockito.mock(ConnectContext.class); Mockito.when(ctx.getQueryIdentifier()).thenReturn("stmt[1, query-id]"); + AtomicBoolean published = new AtomicBoolean(false); UserException thrown = Assertions.assertThrows(UserException.class, () -> Deencapsulation.invoke(command, "publishTheOverwrite", droppedTarget, - Lists.newArrayList("p1"), Lists.newArrayList("tp1"), - new OlapInsertCommandContext(false, true), ctx)); + new OlapInsertCommandContext(false, true), ctx, + (InsertOverwriteTableCommand.OverwritePublication) () -> published.set(true))); Assertions.assertTrue(thrown.getMessage() != null && thrown.getMessage().contains("iot_dropped_target") && thrown.getMessage().contains("was dropped"), "the failure has to name the table whose swap could not be issued, but was: " + thrown.getMessage()); + Assertions.assertFalse(published.get(), + "nothing may be published into a table whose write lock could not be taken"); } }