This is an automated email from the ASF dual-hosted git repository.
yujun777 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new ac78bd1cdae [fix](insert) Finish or fail a cancelled INSERT OVERWRITE
instead of reporting success (#68662)
ac78bd1cdae is described below
commit ac78bd1cdae01d943693071c0245b71308fb9567
Author: yujun <[email protected]>
AuthorDate: Sat Oct 10 11:34:08 2026 +0800
[fix](insert) Finish or fail a cancelled INSERT OVERWRITE instead of
reporting success (#68662)
An `INSERT OVERWRITE` is two-phase: the rows are committed into
temporary partitions first, and a later swap publishes them. A
cancellation that lands between the two halves cannot 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
(`InsertIntoTableCommand` hangs the offsets on the transaction,
`DatabaseTransactionMgr.updateCatalogAfterCommitted` applies them on
commit and replays them). The command used to answer every cancellation
in that window by dropping the temporary partitions and returning
normally, which published nothing while telling the client the overwrite
had succeeded, and left the offsets advanced past rows no target holds
-- so a re-run reads from the advanced offset and that range is silently
missing. That is the general shape of `INSERT OVERWRITE t SELECT ...
FROM stream(...)`, and also of an IVM partition refresh, which resets
the offsets of exactly the partitions it replaces. Cancellations that
landed before the rows were committed were answered the same way:
success for a statement that did nothing.
Fix: decide by whether the insert committed anything.
- If it did not -- a cancellation before the insert ran, or one landing
after an insert that took the path which begins no transaction, because
its plan folded to an empty relation, so the temporary partitions hold
no rows and there is no offset to keep -- nothing is durable, so the
statement fails, like the cancellation the inner insert already reports
for itself, and a re-run reads the same rows.
- If it did, the cancellation is too late: the swap runs, the offset
advance and the publication stay aligned, and the success the statement
reports is the outcome it is. The same decision covers an error response
over committed rows, i.e. a publication that timed out after the commit,
which `insert_visible_timeout_return_mode=error` reports as an error:
the rows exist, so they are published and the client keeps the error.
The decision is taken where the window is, and once more under the
target table's write lock -- the wait for that lock is where a
cancellation would otherwise slip through unseen. It applies to both
routes: the explicit-partition one, and `PARTITION(*)`, whose empty-plan
shape reaches the same no-transaction path when
`enable_strict_consistency_dml` is off. A target dropped under the swap
fails the statement now, instead of acknowledging an overwrite that
never happened.
Three debug points inject a cancellation at each point of the window
(`cancelBeforeTheInsertOfAnOverwrite`,
`cancelBetweenTheTwoHalvesOfAnOverwrite`,
`cancelWhileTheSwapWaitsForTheTableLock`), each scoped by table name, in
the same shape as the existing `failBetweenTheTwoHalvesOfAnOverwrite`.
Scope: this closes the cancellation and the error response, the two
entries into the window that report something other than what happened
to the rows; #68390's per-partition rebuild requirement, raised before a
refresh reads, covers the failure and master-switch halves. Not fixed
here, and tracked in #68679: the FE stopping between the commit and the
swap, a `replacePartition` that throws, a remote commit whose reply is
lost, and an ambiguous cloud commit. All of those need the publication
to become a committed action of the insert transaction (or the
consumption position to move with the publication), rather than another
check in this command.
### Release note
`KILL`/cancellation of an `INSERT OVERWRITE` that lands after its rows
were committed now completes the overwrite instead of silently reporting
a success that published nothing. One that finds nothing committed --
including an overwrite whose plan folded to an empty relation -- reports
an error and leaves the target untouched. An error response over
committed rows, i.e. a publication timeout in error mode, now publishes
them instead of rolling them back. An overwrite whose target is dropped
before its swap fails instead of reporting success.
---
.../AbstractInsertOverwriteManager.java | 13 ++
.../insertoverwrite/InsertOverwriteManager.java | 25 +++
.../RemoteInsertOverwriteManager.java | 13 ++
.../commands/insert/AbstractInsertExecutor.java | 11 +
.../insert/BaseExternalTableInsertExecutor.java | 1 +
.../commands/insert/InsertCommandContext.java | 21 ++
.../commands/insert/InsertIntoTableCommand.java | 4 +
.../insert/InsertOverwriteTableCommand.java | 217 +++++++++++++++---
.../plans/commands/insert/OlapInsertExecutor.java | 4 +
.../commands/insert/RemoteOlapInsertExecutor.java | 1 +
.../insert/InsertOverwriteDroppedTargetTest.java | 73 ++++++
.../test_insert_overwrite_cancel.out | 24 ++
.../test_insert_overwrite_cancel.groovy | 246 +++++++++++++++++++++
13 files changed, 627 insertions(+), 26 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 50321927fc5..04088d48fb6 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 3f0f7feff41..9863a6fdfca 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 class InsertOverwriteManager extends MasterDaemon
implements Writable, Ab
// 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.
+ *
+ * <p>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<Long, Long> relations = partitionPairs.get(groupId);
ArrayList<String> oldNames = new ArrayList<>();
@@ -205,6 +220,16 @@ public class InsertOverwriteManager extends MasterDaemon
implements Writable, Ab
+ "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 4e21501c1ee..65e3f9977fd 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 class RemoteInsertOverwriteManager implements
AbstractInsertOverwriteMana
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/AbstractInsertExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
index bc501b5816b..78d6375aed3 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 abstract class AbstractInsertExecutor {
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<TableStreamUpdateInfo>
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 11d9dd8ee8d..b13027e99ff 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 @@ public abstract class BaseExternalTableInsertExecutor
extends AbstractInsertExec
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 1b4e28117bc..41dce99761f 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,25 @@ package org.apache.doris.nereids.trees.plans.commands.insert;
*/
public abstract class InsertCommandContext {
+ /**
+ * 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.
+ *
+ * <p>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 committed = false;
+
+ public boolean hasCommitted() {
+ return committed;
+ }
+
+ 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 3606ea72526..a8c23b27e5f 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
@@ -745,6 +745,10 @@ public class InsertIntoTableCommand extends Command
AbstractInsertExecutor insertExecutor = initPlan(ctx, executor);
// An empty Table Stream read still needs to commit its offset update
atomically.
if (!insertExecutor.requiresTransaction()) {
+ // 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 4fb6604d930..d6575f407bb 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
@@ -107,6 +107,52 @@ public class InsertOverwriteTableCommand extends Command
public static final String
DEBUG_POINT_FAIL_BETWEEN_THE_HALVES_OF_AN_OVERWRITE =
"InsertOverwriteTableCommand.failBetweenTheTwoHalvesOfAnOverwrite";
+ /**
+ * 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.
+ *
+ * <p>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_BEFORE_THE_INSERT_OF_AN_OVERWRITE =
+ "InsertOverwriteTableCommand.cancelBeforeTheInsertOfAnOverwrite";
+
+ /**
+ * 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.
+ *
+ * <p>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";
+
+ /**
+ * 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";
+
+ /**
+ * 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
+ interface OverwritePublication {
+ void publish() throws UserException;
+ }
+
private static final Logger LOG =
LogManager.getLogger(InsertOverwriteTableCommand.class);
private LogicalPlan originLogicalQuery;
@@ -253,45 +299,77 @@ public class InsertOverwriteTableCommand extends Command
}
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.
- insertIntoAutoDetect(ctx, executor, taskId);
- insertOverwriteManager.taskGroupSuccess(taskId, (OlapTable)
targetTable, isForceDropPartition());
+ InsertCommandContext insertCtx = insertIntoAutoDetect(ctx,
executor, groupId);
+ if (isCancelled.get() && !insertCtx.hasCommitted()) {
+ // 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);
+ }
+ // 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.replacePartitionsOfTaskGroup(groupId, (OlapTable)
targetTable,
+ isForceDropPartition()));
+ insertOverwriteManager.finishTaskGroup(groupId);
} else {
// it's overwrite table(as all partitions) or specific
partition(s)
List<String> tempPartitionNames =
InsertOverwriteUtil.generateTempPartitionNames(partitionNames);
+
cancelTheOverwriteAt(DEBUG_POINT_CANCEL_BEFORE_THE_INSERT_OF_AN_OVERWRITE,
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);
+ InsertCommandContext insertCtx = insertIntoPartitions(ctx,
executor, tempPartitionNames, wholeTable);
+
cancelTheOverwriteAt(DEBUG_POINT_CANCEL_BETWEEN_THE_HALVES_OF_AN_OVERWRITE,
targetTable);
if (isCancelled.get()) {
- LOG.info("insert overwrite is cancelled before
replacePartition, queryId: {}",
- ctx.getQueryIdentifier());
- insertOverwriteManager.taskFail(taskId);
- return;
+ 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
+ // 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());
}
failBetweenTheTwoHalvesOfAnOverwrite(targetTable);
- InsertOverwriteUtil.replacePartition(targetTable,
partitionNames, tempPartitionNames,
- isForceDropPartition());
+ // 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<String> 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());
@@ -300,9 +378,10 @@ public class InsertOverwriteTableCommand extends Command
}
} 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;
@@ -382,12 +461,91 @@ public class InsertOverwriteTableCommand extends Command
throw new UserException("debug point: " +
DEBUG_POINT_FAIL_BETWEEN_THE_HALVES_OF_AN_OVERWRITE);
}
+ /**
+ * 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.
+ *
+ * <p>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 debugPointName, TableIf
targetTable) {
+ if
(!targetTable.getName().equals(DebugPointUtil.getDebugParamOrDefault(
+ debugPointName, "table_name", ""))) {
+ return;
+ }
+ LOG.info("debug point {} cancels the overwrite of {}", debugPointName,
targetTable.getName());
+ cancel();
+ }
+
+ /**
+ * Publishes this overwrite by running its swap, with a last look at the
cancellation flag taken under the
+ * lock the swap contends for.
+ *
+ * <p>{@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 an empty result.
+ *
+ * <p>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, InsertCommandContext
insertCtx, ConnectContext ctx,
+ OverwritePublication publication) throws UserException {
+ if (!(targetTable instanceof OlapTable) || targetTable instanceof
RemoteOlapTable) {
+ publication.publish();
+ return;
+ }
+ OlapTable olapTable = (OlapTable) targetTable;
+ if (!olapTable.writeLockIfExist()) {
+ // 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);
+ if (isCancelled.get() && !insertCtx.hasCommitted()) {
+ throw cancelledBeforeTheRowsWereCommitted("while the swap
waited for the table lock", ctx);
+ }
+ publication.publish();
+ } 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
+ * 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: "
+ + ctx.getQueryIdentifier());
+ }
+
private void runInsertCommand(LogicalPlan logicalQuery,
InsertCommandContext insertCtx,
ConnectContext ctx, StmtExecutor executor) throws Exception {
InsertIntoTableCommand insertCommand = new
InsertIntoTableCommand(logicalQuery, labelName,
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);
@@ -401,9 +559,11 @@ public class InsertOverwriteTableCommand extends Command
* @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<String> tempPartitionNames,
- boolean wholeTable)
+ private InsertCommandContext insertIntoPartitions(ConnectContext ctx,
StmtExecutor executor,
+ List<String> tempPartitionNames, boolean wholeTable)
throws Exception {
// copy sink tot replace by tempPartitions
UnboundLogicalSink<?> copySink;
@@ -448,6 +608,7 @@ public class InsertOverwriteTableCommand extends Command
throw new UserException("Current catalog does not support insert
overwrite yet.");
}
runInsertCommand(copySink, insertCtx, ctx, executor);
+ return insertCtx;
}
/**
@@ -455,8 +616,11 @@ public class InsertOverwriteTableCommand extends Command
*
* @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) {
@@ -469,6 +633,7 @@ public class InsertOverwriteTableCommand extends Command
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/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 29b80a91aad..70387f50619 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 @@ public class OlapInsertExecutor extends
AbstractInsertExecutor {
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 c95c3b0d71d..511a0a5bd4e 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 @@ public class RemoteOlapInsertExecutor extends
OlapInsertExecutor {
} 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/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 00000000000..ac1dd5e54c2
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteDroppedTargetTest.java
@@ -0,0 +1,73 @@
+// 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 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.
+ *
+ * <p>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.
+ *
+ * <p>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, and the
+ * publication it is handed records whether it was reached at all.
+ */
+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]");
+ AtomicBoolean published = new AtomicBoolean(false);
+
+ UserException thrown = Assertions.assertThrows(UserException.class,
+ () -> Deencapsulation.invoke(command, "publishTheOverwrite",
droppedTarget,
+ 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");
+ }
+}
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 00000000000..90d04658c0a
--- /dev/null
+++ b/regression-test/data/insert_overwrite_p0/test_insert_overwrite_cancel.out
@@ -0,0 +1,24 @@
+-- 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
+
+-- !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_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
+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
new file mode 100644
index 00000000000..f83314073dd
--- /dev/null
+++
b/regression-test/suites/insert_overwrite_p0/test_insert_overwrite_cancel.groovy
@@ -0,0 +1,246 @@
+// 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 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.
+ *
+ * <p>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; 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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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
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"""
+ 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 (
+ 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)"""
+
+ // 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(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(beforePoint)
+ }
+ 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(betweenPoint,
+ [table_name: "test_iot_cancel_dst"])
+ sql """INSERT OVERWRITE TABLE test_iot_cancel_dst SELECT * FROM
test_iot_cancel_src"""
+ } finally {
+ 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"""
+
+ // 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(waitsPoint,
+ [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 while the swap waited for
the table lock"
+ }
+ } finally {
+ GetDebugPoint().disableDebugPointForAllFEs(waitsPoint)
+ }
+ 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
+ // 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
+ }
+ 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"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
