diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 18aca01243908..8845eeebaa794 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -1430,4 +1430,11 @@ private ProcedureMessages() {} "DataNode setSystemStatus RPC failed: %s: %s: %s; "; public static final String EXCEPTION_CONFIGNODE_TRYSETNODESTATUS_FAILED_ARG_ARG_ARG_C9BA78F4 = "ConfigNode trySetNodeStatus failed: %s: %s: %s; "; + public static final String + LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9 = + "Procedure {} is past its commit point; keeping every state it executed."; + public static final String LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354 = + "finish procedure without rollback"; + public static final String LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1 = + "delete child procedures without rollback"; } diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 14e52d72b9efe..76a88916a180d 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -1515,4 +1515,10 @@ private ProcedureMessages() {} "DataNode setSystemStatus RPC 失败：%s：%s：%s； "; public static final String EXCEPTION_CONFIGNODE_TRYSETNODESTATUS_FAILED_ARG_ARG_ARG_C9BA78F4 = "ConfigNode trySetNodeStatus 失败：%s：%s：%s； "; + public static final String + LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9 = + "流程 {} 已越过提交点，保留它执行过的所有状态。"; + public static final String LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354 = "结束流程（不回滚）"; + public static final String LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1 = + "删除子流程（不回滚）"; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java index 10ae561c2af57..dc26ced3b6462 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java @@ -99,6 +99,24 @@ public final boolean hasLock() { protected abstract void rollback(Env env) throws IOException, InterruptedException, ProcedureException; + /** + * Whether the states this procedure has already executed may still be undone. The executor asks + * this once, before it starts to roll back the root procedure stack. A procedure which returns + * false is finished without undoing any of its states, see {@link #abandonRollback()}. + */ + protected boolean isRollbackSupported() { + return true; + } + + /** + * Drops the bookkeeping of the states this procedure has executed. The persisted stack indexes + * are what a recovered rollback rebuilds its stack from, so dropping them is what makes the + * decision survive a leader switch. + */ + protected synchronized void abandonRollback() { + stackIndexes = null; + } + public void serialize(DataOutputStream stream) throws IOException { // procid stream.writeLong(this.procId); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java index 82b217b93eab6..0b5460486b29c 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java @@ -40,8 +40,10 @@ import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.Deque; import java.util.HashSet; +import java.util.IdentityHashMap; import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -370,7 +372,7 @@ public boolean removeInternalProcedure(InternalProcedure internalProcedure) { * * @param proc procedure */ - private void executeProcedure(Procedure proc) { + private void executeProcedure(Procedure proc) throws InterruptedException { if (proc.isFinished()) { LOG.debug(ProcedureMessages.IS_ALREADY_FINISHED, proc); return; @@ -689,7 +691,7 @@ private void yieldProcedure(Procedure proc) { * @return lock state */ private LockStateResult executeRootStackRollback( - Long rootProcId, RootProcedureStack procedureStack) { + Long rootProcId, RootProcedureStack procedureStack) throws InterruptedException { Procedure rootProcedure = procedures.get(rootProcId); ProcedureException exception = rootProcedure.getException(); if (exception == null) { @@ -700,14 +702,33 @@ private LockStateResult executeRootStackRollback( () -> store.update(rootProcedure), "root procedure rollback"); } List> subprocStack = procedureStack.getSubproceduresStack(); + // Decide which procedures must keep the states they have executed before anything is rolled + // back: the decision reads the state history, which the rollback itself shrinks. Identity is + // used because procedures override equals with the data of the operation they run. + final Set> proceduresKeepingTheirStates = + Collections.newSetFromMap(new IdentityHashMap<>()); + for (Procedure procedure : subprocStack) { + // A null slot is a procedure whose record was deleted before the snapshot this stack was + // rebuilt from, so there is nothing left to roll back for it. + if (procedure != null && !procedure.isSuccess() && !procedure.isRollbackSupported()) { + proceduresKeepingTheirStates.add(procedure); + } + } int stackTail = subprocStack.size(); while (stackTail-- > 0) { Procedure procedure = subprocStack.get(stackTail); + if (procedure == null) { + continue; + } if (procedure.isSuccess()) { subprocStack.remove(stackTail); cleanupAfterRollback(procedure); continue; } + if (proceduresKeepingTheirStates.contains(procedure)) { + // Finished below, once the rest of the stack has been rolled back. + continue; + } ProcedureLockState lockState = acquireLock(procedure); if (lockState != ProcedureLockState.LOCK_ACQUIRED) { return new LockStateResult<>(lockState, procedure); @@ -730,12 +751,109 @@ private LockStateResult executeRootStackRollback( } } + if (!proceduresKeepingTheirStates.isEmpty()) { + // The children go first: releasing a child's lock writes its record back, which has to be + // covered by the root's cleanup. The root is finished last so that no write of a child + // follows the write which persists the root as finished. + for (Procedure procedure : proceduresKeepingTheirStates) { + if (procedure.hasParent()) { + finishWithoutRollback(procedure); + } + } + for (Procedure procedure : proceduresKeepingTheirStates) { + if (!procedure.hasParent()) { + finishWithoutRollback(procedure); + } + } + subprocStack.removeIf(proceduresKeepingTheirStates::contains); + } + LOG.info( ProcedureMessages.ROLLED_BACK_TIME_DURATION_IS, rootProcedure, rootProcedure.elapsedTime()); rootProcedureCleanup(rootProcedure); return new LockStateResult<>(ProcedureLockState.LOCK_ACQUIRED, rootProcedure); } + /** + * Finishes a procedure without undoing any of the states it has executed. The failure is still + * reported, as the operation the procedure was part of did not complete; what is dropped is only + * the bookkeeping of the rollback, so that neither this rollback nor the one a new leader + * recovers touches this procedure again. + */ + private void finishWithoutRollback(final Procedure procedure) throws InterruptedException { + // The unlock persists the procedure through doReleaseLock(). It is done first, while the + // procedure is neither marked finished nor stripped of its state history, so that a snapshot + // taken right after this write is still a state recovery resumes. Forcing the release matters + // because the procedure is not finished yet, so releaseLock() would otherwise keep the lock of + // procedures which hold it across steps. + releaseLockIfHeld(procedure); + if (procedure.hasParent()) { + // Mark the procedure finished before dropping its state history, so that a worker which + // picks it up concurrently returns at the isFinished() check instead of rolling it back. + if (!procedure.isSuccess()) { + procedure.setState(ProcedureState.ROLLEDBACK); + } + procedure.abandonRollback(); + try { + store.delete(procedure.getProcId()); + // do not remove this procedure when exception occurred + procedures.remove(procedure.getProcId()); + } catch (final Exception e) { + LOG.warn(ProcedureMessages.FAILED_TO_DELETE_PROCEDURE_ON_ROLLBACK, e); + } + } else { + // A child which never executed has no rollback step and is not visited above, so its record + // goes with the ones of the children which were. + // Every child record has to be gone before the root may be persisted as finished: a snapshot + // holding a finished root beside a child record makes recover() fail on that child. Deleting + // is idempotent, so it is retried until durable. + final long[] childProcIds = rollbackStack.get(procedure.getProcId()).getSubprocedureIds(); + for (final long childProcId : childProcIds) { + RetryUtils.executeWithEndlessBackoffRetry( + () -> store.delete(childProcId), + ProcedureMessages.LOG_DELETE_CHILD_PROCEDURES_WITHOUT_ROLLBACK_6E62D1C1); + // The retry helper also returns on interruption. Abort the entire rollback before + // marking the root finished or letting the caller clean it up. + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException(); + } + } + if (!procedure.isSuccess()) { + procedure.setState(ProcedureState.ROLLEDBACK); + } + procedure.abandonRollback(); + // The stack indexes and the states just dropped are persisted here, so that the rollback + // recovered by a new leader leaves this procedure out. Endless retry since this step is not + // idempotent. + RetryUtils.executeWithEndlessBackoffRetry( + () -> store.update(procedure), + ProcedureMessages.LOG_FINISH_PROCEDURE_WITHOUT_ROLLBACK_41F7A354); + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException(); + } + } + procedure.updateMetricsOnFinish(getEnvironment(), procedure.elapsedTime(), false); + LOG.info( + ProcedureMessages + .LOG_PROCEDURE_ARG_IS_PAST_ITS_COMMIT_POINT_KEEPING_EVERY_STATE_IT_EXECUTED_CA21D7E9, + procedure); + } + + /** + * The rollback loop skipped acquireLock() for procedures finished without rollback, so a lock + * restored by restoreLock() on load is still recorded in lockedWhenLoading instead of locked. + * Finish that bookkeeping and release the lock here: rootProcedureCleanup() releases only the + * locks that hasLock() reports, and a procedure finished without rollback must not keep holding + * its lock. + */ + private void releaseLockIfHeld(final Procedure procedure) { + if (procedure.hasLock() || procedure.isLockedWhenLoading()) { + if (acquireLock(procedure) == ProcedureLockState.LOCK_ACQUIRED) { + releaseLock(procedure, true); + } + } + } + private ProcedureLockState acquireLock(Procedure proc) { if (proc.hasLock()) { return ProcedureLockState.LOCK_ACQUIRED; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/RootProcedureStack.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/RootProcedureStack.java index 1fa240997f764..db38c6175bb08 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/RootProcedureStack.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/RootProcedureStack.java @@ -89,7 +89,9 @@ protected synchronized List> getSubproceduresStack() { protected synchronized ProcedureException getException() { if (subprocStack != null) { for (Procedure proc : subprocStack) { - if (proc.hasException()) { + // Null slots come from procedures whose records were deleted before the stack was + // rebuilt, see ProcedureExecutor#executeRootStackRollback. + if (proc != null && proc.hasException()) { return proc.getException(); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java index f34225cede983..5f518f728e936 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java @@ -31,6 +31,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Iterator; import java.util.List; import java.util.concurrent.ConcurrentLinkedDeque; @@ -252,12 +253,49 @@ protected boolean isEofState() { return !states.isEmpty() && states.getLast() == EOF_STATE; } + @Override + protected synchronized void abandonRollback() { + super.abandonRollback(); + states.clear(); + } + /** - * Used by the default implementation of abort() to know if the current state can be aborted and - * rollback can be triggered. + * Whether a failure in the specified state can be rolled back. A procedure asks this in its error + * handling to tell a fatal failure from a retriable one; the executor asks it once, through the + * overload without a state, to decide whether the procedure may be rolled back at all. + * + *

The executor asks with the state the procedure is at: the state which failed or is + * scheduled, or the last state executed when every state has already run. A procedure which + * returns false for the states after its commit point is therefore not rolled back once that + * point has been passed. */ protected boolean isRollbackSupported(final TState state) { - return false; + return true; + } + + /** + * Whether the states this procedure has executed may still be rolled back. The answer comes from + * the state the procedure is at, through the per-state overload above. + */ + @Override + protected final boolean isRollbackSupported() { + return isRollbackSupported(getRollbackDecisionState()); + } + + /** + * The state a rollback decision is taken at. The last element of the history is the state which + * failed or is scheduled; once every state has run, the end marker is last and the decision falls + * on the state before it. + */ + private TState getRollbackDecisionState() { + final Iterator stateIds = states.descendingIterator(); + while (stateIds.hasNext()) { + final int stateId = stateIds.next(); + if (stateId != EOF_STATE) { + return getState(stateId); + } + } + return getInitialState(); } private boolean noMoreState() { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveAINodeProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveAINodeProcedure.java index 23b041c2587e9..a6fdd7d61517e 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveAINodeProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveAINodeProcedure.java @@ -135,6 +135,15 @@ protected void rollbackState( // no need to rollback } + /** + * A failure in either state is retried instead of failing the procedure, which has nothing to + * roll back. + */ + @Override + protected boolean isRollbackSupported(final RemoveAINodeState state) { + return false; + } + @Override protected RemoveAINodeState getState(int stateId) { return RemoveAINodeState.values()[stateId]; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java index be6f67007a3db..a354f0ff3a393 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractSetPropertiesProcedure.java @@ -162,6 +162,23 @@ private void setProperties(final ConfigNodeProcedureEnv env) { protected abstract ConfigPhysicalPlan createSetPropertiesPlan( Map properties, boolean isRollback); + /** + * The properties are set on the ConfigNode once SET_PROPERTIES has completed, and the ConfigNode + * schema is the source of truth the DataNodes resync from. The procedure therefore stays + * rollbackable up to SET_PROPERTIES and not after it. + */ + @Override + protected boolean isRollbackSupported(final SetTablePropertiesState state) { + switch (state) { + case VALIDATE_TABLE: + case PRE_RELEASE: + case SET_PROPERTIES: + return true; + default: + return false; + } + } + @Override protected void rollbackState( final ConfigNodeProcedureEnv env, final SetTablePropertiesState state) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java index 44f9e2f07f9fe..07bdbf44c55c9 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java @@ -154,6 +154,23 @@ protected String getActionMessage() { return "add table column"; } + /** + * The column is part of the table on the ConfigNode once ADD_COLUMN has completed, and the + * ConfigNode schema is the source of truth the DataNodes resync from. The procedure therefore + * stays rollbackable up to ADD_COLUMN and not after it. + */ + @Override + protected boolean isRollbackSupported(final AddTableColumnState state) { + switch (state) { + case COLUMN_CHECK: + case PRE_RELEASE: + case ADD_COLUMN: + return true; + default: + return false; + } + } + @Override protected void rollbackState(final ConfigNodeProcedureEnv env, final AddTableColumnState state) throws IOException, InterruptedException, ProcedureException { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java index 12491f24a4867..1cfda63fcf736 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java @@ -161,6 +161,7 @@ private void alterColumnDataType(final ConfigNodeProcedureEnv env) { setNextState(AlterTableColumnDataTypeState.COMMIT_RELEASE); } + /** The rollback of this procedure is a no-op, so it is never rolled back. */ @Override protected boolean isRollbackSupported(final AlterTableColumnDataTypeState state) { return false; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java index f2379ae1256eb..ac573463ebb23 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java @@ -212,7 +212,15 @@ private void commitReleaseTable(final ConfigNodeProcedureEnv env) { @Override protected boolean isRollbackSupported(final CreateTableState state) { - return true; + switch (state) { + case CHECK_TABLE_EXISTENCE: + case PRE_CREATE: + case PRE_RELEASE: + case COMMIT_CREATE: + return true; + default: + return false; + } } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java index a51c636551ea7..75000e36ef67a 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java @@ -233,6 +233,7 @@ private void dropColumn(final ConfigNodeProcedureEnv env) { } } + /** The rollback of this procedure is a no-op, so it is never rolled back. */ @Override protected boolean isRollbackSupported(final DropTableColumnState state) { return false; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java index cdb4d4b71c81d..54217212a1151 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java @@ -386,9 +386,22 @@ private void dropTable(final ConfigNodeProcedureEnv env) { } } + /** + * The data is deleted logically once DELETE_DATA has completed: the deletion is applied to every + * data region, and only the physical removal of the mod files continues in the background. The + * procedure is therefore not rolled back once it has moved on to DELETE_DEVICES. A view holds no + * data and goes straight to DROP_TABLE, see {@link DropViewProcedure}. + */ @Override protected boolean isRollbackSupported(final DropTableState state) { - return state == DropTableState.CHECK_AND_INVALIDATE_TABLE || state == DropTableState.PRE_DELETE; + switch (state) { + case CHECK_AND_INVALIDATE_TABLE: + case PRE_DELETE: + case DELETE_DATA: + return true; + default: + return false; + } } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java index 11d10901a352a..03088935ce616 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java @@ -154,6 +154,23 @@ private void renameColumn(final ConfigNodeProcedureEnv env) { } } + /** + * The column is renamed on the ConfigNode once RENAME_COLUMN has completed, and the ConfigNode + * schema is the source of truth the DataNodes resync from. The procedure therefore stays + * rollbackable up to RENAME_COLUMN and not after it. + */ + @Override + protected boolean isRollbackSupported(final RenameTableColumnState state) { + switch (state) { + case COLUMN_CHECK: + case PRE_RELEASE: + case RENAME_COLUMN: + return true; + default: + return false; + } + } + @Override protected void rollbackState(final ConfigNodeProcedureEnv env, final RenameTableColumnState state) throws IOException, InterruptedException, ProcedureException { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java index e40b041d0f6e7..5296253304913 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableProcedure.java @@ -144,6 +144,23 @@ private void renameTable(final ConfigNodeProcedureEnv env) { } } + /** + * The table is renamed on the ConfigNode once RENAME_TABLE has completed, and the ConfigNode + * schema is the source of truth the DataNodes resync from. The procedure therefore stays + * rollbackable up to RENAME_TABLE and not after it. + */ + @Override + protected boolean isRollbackSupported(final RenameTableState state) { + switch (state) { + case COLUMN_CHECK: + case PRE_RELEASE: + case RENAME_TABLE: + return true; + default: + return false; + } + } + @Override protected void rollbackState(final ConfigNodeProcedureEnv env, final RenameTableState state) throws IOException, InterruptedException, ProcedureException { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java index 73a7732fc87ea..163150c0a09e4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/view/DropViewProcedure.java @@ -20,6 +20,7 @@ package org.apache.iotdb.confignode.procedure.impl.schema.table.view; import org.apache.iotdb.confignode.procedure.impl.schema.table.DropTableProcedure; +import org.apache.iotdb.confignode.procedure.state.schema.DropTableState; import org.apache.iotdb.confignode.procedure.store.ProcedureType; import java.io.DataOutputStream; @@ -38,6 +39,23 @@ public DropViewProcedure( super(database, tableName, queryId, isGeneratedByPipe); } + /** + * A view holds no data, so the procedure goes straight from PRE_DELETE to DROP_TABLE. The view is + * removed from the ConfigNode once DROP_TABLE has completed; before that it can still be + * restored, so the procedure is not rolled back from COMMIT_DELETE on. + */ + @Override + protected boolean isRollbackSupported(final DropTableState state) { + switch (state) { + case CHECK_AND_INVALIDATE_TABLE: + case PRE_DELETE: + case DROP_TABLE: + return true; + default: + return false; + } + } + @Override public void serialize(final DataOutputStream stream) throws IOException { stream.writeShort( diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java new file mode 100644 index 0000000000000..340f1b88c5cda --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedureTest.java @@ -0,0 +1,185 @@ +/* + * 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.iotdb.confignode.procedure.impl; + +import org.apache.iotdb.confignode.procedure.entity.SimpleSTMProcedure; +import org.apache.iotdb.confignode.procedure.env.TestProcEnv; +import org.apache.iotdb.confignode.procedure.exception.ProcedureException; + +import org.junit.Assert; +import org.junit.Test; + +public class StateMachineProcedureTest { + + private enum State { + PREPARE, + APPLY, + RELEASE + } + + /** + * A procedure whose commit point is the completion of {@code APPLY}: it stays rollbackable while + * it is at {@code PREPARE} or {@code APPLY}, and not once it has moved on to {@code RELEASE}. + */ + private static class CommitPointProcedure extends StateMachineProcedure { + + private boolean failAtApply; + private boolean failAtRelease; + + @Override + protected Flow executeFromState(final TestProcEnv env, final State state) { + switch (state) { + case PREPARE: + setNextState(State.APPLY); + break; + case APPLY: + if (failAtApply) { + setFailure(new ProcedureException("apply failed")); + } else { + setNextState(State.RELEASE); + } + break; + case RELEASE: + if (failAtRelease) { + setFailure(new ProcedureException("release failed")); + } + return Flow.NO_MORE_STATE; + default: + throw new IllegalStateException(); + } + return Flow.HAS_MORE_STATE; + } + + @Override + protected void rollbackState(final TestProcEnv env, final State state) { + // The point of the test is that this is never called once the commit point has been passed. + } + + @Override + protected boolean isRollbackSupported(final State state) { + return state != State.RELEASE; + } + + @Override + protected State getState(final int stateId) { + return State.values()[stateId]; + } + + @Override + protected int getStateId(final State state) { + return state.ordinal(); + } + + @Override + protected State getInitialState() { + return State.PREPARE; + } + + /** Runs one state, the way the executor does. */ + private void runStep() throws InterruptedException { + execute(new TestProcEnv()); + } + + private State currentState() { + return getCurrentState(); + } + + private void addSlot(final int index) { + addStackIndex(index); + } + + private int[] stackIndexes() { + return getStackIndexes(); + } + + private void abandon() { + abandonRollback(); + } + } + + @Test + public void rollbackIsAllowedUntilTheStateAfterTheCommitPointIsReached() + throws InterruptedException { + final CommitPointProcedure procedure = new CommitPointProcedure(); + + // PREPARE has completed and APPLY is scheduled. + procedure.runStep(); + Assert.assertEquals(State.APPLY, procedure.currentState()); + Assert.assertTrue(procedure.isRollbackSupported()); + + // APPLY has completed: the procedure is at RELEASE and must not be rolled back any more. + procedure.runStep(); + Assert.assertEquals(State.RELEASE, procedure.currentState()); + Assert.assertFalse(procedure.isRollbackSupported()); + + // Every state has run; the decision falls on the last state which ran, not on the end marker. + procedure.runStep(); + Assert.assertFalse(procedure.isRollbackSupported()); + } + + @Test + public void aFailureBeforeTheCommitPointStillAllowsTheRollback() throws InterruptedException { + final CommitPointProcedure procedure = new CommitPointProcedure(); + procedure.runStep(); + + procedure.failAtApply = true; + procedure.runStep(); + + Assert.assertEquals(State.APPLY, procedure.currentState()); + Assert.assertTrue(procedure.isRollbackSupported()); + } + + @Test + public void aFailureAfterTheCommitPointDoesNotAllowTheRollback() throws InterruptedException { + final CommitPointProcedure procedure = new CommitPointProcedure(); + procedure.runStep(); + procedure.runStep(); + + procedure.failAtRelease = true; + procedure.runStep(); + + Assert.assertFalse(procedure.isRollbackSupported()); + } + + @Test + public void abandoningTheRollbackDropsTheStackSlotsAndTheStateHistory() + throws InterruptedException { + final CommitPointProcedure procedure = new CommitPointProcedure(); + procedure.runStep(); + procedure.runStep(); + procedure.addSlot(0); + procedure.addSlot(1); + procedure.addSlot(2); + Assert.assertEquals(3, procedure.stackIndexes().length); + Assert.assertFalse(procedure.isRollbackSupported()); + + procedure.abandon(); + + // A recovered rollback rebuilds its stack from the persisted indexes, so they must be gone. + Assert.assertNull(procedure.stackIndexes()); + // An empty history reports the initial state, which is how a drained history shows up. + Assert.assertEquals(State.PREPARE, procedure.currentState()); + } + + @Test + public void aProcedureWhichOverridesNothingIsRollbackableEverywhere() { + Assert.assertTrue(new SimpleSTMProcedure().isRollbackSupported()); + } +}