This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 4473a133d08 Fix table rename on a DataNode that missed the pre-release
(#18800)
4473a133d08 is described below
commit 4473a133d08490e94bfc21e4227f949cf0158dc6
Author: alpass163 <[email protected]>
AuthorDate: Fri Oct 9 09:14:03 2026 +0800
Fix table rename on a DataNode that missed the pre-release (#18800)
---
.../iotdb/confignode/i18n/ProcedureMessages.java | 2 +
.../iotdb/confignode/i18n/ProcedureMessages.java | 11 ++--
.../procedure/impl/schema/SchemaUtils.java | 17 +++++--
.../table/AbstractAlterOrDropTableProcedure.java | 19 +++++++
.../impl/schema/table/RenameTableProcedure.java | 2 +-
.../db/schemaengine/table/DataNodeTableCache.java | 27 +++++++++-
.../schemaengine/table/DataNodeTableCacheTest.java | 59 ++++++++++++++++++++++
7 files changed, 125 insertions(+), 12 deletions(-)
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 e04460375f9..94abb3faab4 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
@@ -1200,6 +1200,8 @@ public final class ProcedureMessages {
public static final String
LOG_FINISHED_SUBPROCEDURE_PID_ARG_RESUME_PROCESSING_PPID_ARG_93ED990B =
"Finished subprocedure pid={}, resume processing ppid={}";
public static final String LOG_HALT_PID_ARG_ACTIVECOUNT_ARG_411F3EBF = "Halt
pid={}, activeCount={}";
public static final String
LOG_EXCEPTION_HAPPENED_WORKER_ARG_EXECUTE_PROCEDURE_ARG_6E3AD27D = "Exception
happened when worker {} execute procedure {}";
+ public static final String
LOG_FAILED_TO_COMMIT_RELEASE_ARG_FOR_TABLE_ARG_ARG_TO_DATANODE_FAILURE_RESULTS_ARG_55386572
=
+ "Failed to commit-release {} for table {}.{} to DataNode, failure
results: {}";
public static final String LOG_WORKER_STUCK_ARG_ARG_RUN_TIME_ARG_MS_FB612354
= "Worker stuck {}({}), run time {} ms";
public static final String
LOG_PROCEDURE_WORKERS_ARG_RUNNING_ARG_RUNNING_STUCK_1565936D = "Procedure
workers: {} is running, {} is running and stuck";
public static final String
LOG_ADD_PROCEDURE_ARG_AS_ARG_TH_ROLLBACK_STEP_C71B2184 = "Add procedure {} as
the {}th rollback step";
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 3014e1c0076..d4d5dc2770a 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
@@ -1137,6 +1137,9 @@ public final class ProcedureMessages {
"暂停 pid={}, activeCount={}";
public static final String
LOG_EXCEPTION_HAPPENED_WORKER_ARG_EXECUTE_PROCEDURE_ARG_6E3AD27D =
"worker {} 执行 procedure {} 时发生异常";
+ public static final String
+
LOG_FAILED_TO_COMMIT_RELEASE_ARG_FOR_TABLE_ARG_ARG_TO_DATANODE_FAILURE_RESULTS_ARG_55386572
=
+ "在 {} procedure 操作中， 对表{}.{} 的正式提交失败，失败结果：{}";
public static final String LOG_WORKER_STUCK_ARG_ARG_RUN_TIME_ARG_MS_FB612354
=
"Worker 卡住 {}({})，运行时间 {} ms";
public static final String
LOG_PROCEDURE_WORKERS_ARG_RUNNING_ARG_RUNNING_STUCK_1565936D =
@@ -1500,11 +1503,9 @@ public final class ProcedureMessages {
public static final String
MESSAGE_THE_PREVIOUS_ATTEMPT_FAILED_WITH_ARG_AND_THIS_STATE_IS_BEING_RETRIED_7A541F27
=
"上一次尝试因“%s”失败，正在重试此状态。";
- public static final String
- MESSAGE_THE_STATE_FAILED_WITH_ARG_AND_ROLLBACK_IS_PENDING_E7B43829 =
- "此状态因“%s”失败，正在等待回滚。";
- public static final String MESSAGE_ROLLING_BACK_AFTER_FAILURE_ARG_474DF456 =
- "正在回滚，失败原因：%s。";
+ public static final String
MESSAGE_THE_STATE_FAILED_WITH_ARG_AND_ROLLBACK_IS_PENDING_E7B43829 =
+ "此状态因“%s”失败，正在等待回滚。";
+ public static final String MESSAGE_ROLLING_BACK_AFTER_FAILURE_ARG_474DF456 =
"正在回滚，失败原因：%s。";
public static final String
MESSAGE_ROLLING_BACK_AFTER_AN_EARLIER_FAILURE_850D0AF5 =
"正在回滚此前发生的失败。";
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java
index 69a3cec4400..ea91664834b 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java
@@ -288,11 +288,9 @@ public class SchemaUtils {
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
- public static Map<Integer, TSStatus> commitReleaseTable(
- final String database,
- final String tableName,
- final ConfigManager configManager,
- final @Nullable String oldName) {
+ /** Build the COMMIT_UPDATE_TABLE request used to commit-release a table
change to DataNodes. */
+ public static TUpdateTableReq buildCommitUpdateTableReq(
+ final String database, final String tableName, final @Nullable String
oldName) {
final TUpdateTableReq req = new TUpdateTableReq();
req.setType(TsTableInternalRPCType.COMMIT_UPDATE_TABLE.getOperationType());
final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
@@ -304,6 +302,15 @@ public class SchemaUtils {
}
req.setTableInfo(outputStream.toByteArray());
req.setOldName(oldName);
+ return req;
+ }
+
+ public static Map<Integer, TSStatus> commitReleaseTable(
+ final String database,
+ final String tableName,
+ final ConfigManager configManager,
+ final @Nullable String oldName) {
+ final TUpdateTableReq req = buildCommitUpdateTableReq(database, tableName,
oldName);
final DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler
=
new DataNodeAsyncRequestContext<>(
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractAlterOrDropTableProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractAlterOrDropTableProcedure.java
index 76dfce38808..b4aab9949c2 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractAlterOrDropTableProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractAlterOrDropTableProcedure.java
@@ -136,6 +136,25 @@ public abstract class AbstractAlterOrDropTableProcedure<T>
}
}
+ protected void commitReleaseWithHA(
+ final ConfigNodeProcedureEnv env, final @Nullable String oldName) {
+ final TUpdateTableReq req =
+ SchemaUtils.buildCommitUpdateTableReq(database, table.getTableName(),
oldName);
+ final boolean proceeded =
+ new
ClusterCachePropagator(SchemaUtils.filterFencedDataNode(env.getConfigManager()))
+ .propagate(targets -> SchemaUtils.broadcastTableUpdate(req,
targets));
+
+ if (!proceeded) {
+ LOGGER.warn(
+ ProcedureMessages
+
.LOG_FAILED_TO_COMMIT_RELEASE_ARG_FOR_TABLE_ARG_ARG_TO_DATANODE_FAILURE_RESULTS_ARG_55386572,
+ getActionMessage(),
+ database,
+ table.getTableName(),
+ ProcedureMessages.FAILED_TO_PROVE_DN_IS_FENCED);
+ }
+ }
+
@Override
protected boolean isRollbackSupported(final T state) {
return true;
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 ba5c379275b..e40b041d0f6 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
@@ -85,7 +85,7 @@ public class RenameTableProcedure extends
AbstractAlterOrDropTableProcedure<Rena
ProcedureMessages.COMMIT_RELEASE_INFO_OF_TABLE_WHEN_RENAMING_TABLE,
database,
tableName);
- commitRelease(env, tableName);
+ commitReleaseWithHA(env, tableName);
return Flow.NO_MORE_STATE;
default:
setFailure(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
index 98ca71e0eac..525bf511f4e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
@@ -329,7 +329,7 @@ public class DataNodeTableCache implements ITableCache {
database,
tableName);
if (Objects.nonNull(oldName)) {
- removeTableFromSpecialStatusMap(database, oldName);
+ repairMissedRenamePreUpdate(database, tableName, oldName);
}
return;
}
@@ -370,6 +370,31 @@ public class DataNodeTableCache implements ITableCache {
}
}
+ /**
+ * Repairs the cache after this DataNode missed the pre-update of a rename
procedure. The
+ * ConfigNode has already renamed the table, so the new name can only be
re-fetched, and the old
+ * name must not stay in the cache as a usable table.
+ */
+ private void repairMissedRenamePreUpdate(
+ final String database, final String newName, final String oldName) {
+ removeTableFromSpecialStatusMap(database, oldName);
+ // drop the old table
+ databaseTableMap.computeIfPresent(
+ database,
+ (databaseName, tables) -> {
+ tables.remove(oldName);
+ return tables;
+ });
+ // A placeholder makes the next lookup of the new name fetch its
definition from the ConfigNode,
+ // which holds the committed rename.
+ if (Objects.isNull(getTableFromCache(database, newName))) {
+ specialStatusMap
+ .computeIfAbsent(database, k -> new ConcurrentHashMap<>())
+ .put(newName, new Pair<>(new NonCommittableTsTable(newName), 0L));
+ }
+ instanceVersion.incrementAndGet();
+ }
+
private void commitDeleteTable(String database, final String tableName) {
if (databaseTableMap.containsKey(database)) {
databaseTableMap.get(database).remove(tableName);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
index d8d1eddafe4..bec226b21ab 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
@@ -21,14 +21,18 @@ package org.apache.iotdb.db.schemaengine.table;
import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.schema.table.NonCommittableTsTable;
import org.apache.iotdb.commons.schema.table.PreDeleteTsTable;
import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil;
import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema;
+import org.apache.iotdb.commons.utils.PathUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.utils.Pair;
import org.junit.Assert;
import org.junit.Test;
@@ -37,7 +41,9 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.Semaphore;
public class DataNodeTableCacheTest {
@@ -120,6 +126,33 @@ public class DataNodeTableCacheTest {
}
}
+ @Test
+ public void commitRenameRepairsCacheWhenPreReleaseWasMissed() throws
Exception {
+ final ITableCache cache = DataNodeTableCache.getInstance();
+ final String oldName = "table1";
+ final String newName = "table2";
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ try {
+ // The activation snapshot of a DataNode that started during the
pre-release: the old name is
+ // a using table, and it is also marked as NonCommittable.
+ final Map<String, List<TsTable>> usingTables = new HashMap<>();
+ usingTables.put(TABLE_CACHE_TEST_DATABASE,
Collections.singletonList(createTable(oldName)));
+ final Map<String, List<TsTable>> specialTables = new HashMap<>();
+ specialTables.put(
+ TABLE_CACHE_TEST_DATABASE, Collections.singletonList(new
NonCommittableTsTable(oldName)));
+ cache.init(
+ TsTableInternalRPCUtil.serializeTableInitializationInfo(usingTables,
specialTables));
+
+ // The commit is the only sign of the rename, and the HA broadcast may
send it more than once.
+ cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, newName, oldName);
+ assertRenameRepaired(cache, oldName, newName);
+ cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, newName, oldName);
+ assertRenameRepaired(cache, oldName, newName);
+ } finally {
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ }
+ }
+
@Test
public void fetchingPreDeleteTableReportsDedicatedStatusCode() throws
Exception {
final ITableCache cache = DataNodeTableCache.getInstance();
@@ -166,6 +199,32 @@ public class DataNodeTableCacheTest {
cache, fetchedTables, DATABASE, tableName,
LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
}
+ /** The old name must be gone, and the new name left as a placeholder to be
re-fetched. */
+ private static void assertRenameRepaired(
+ final ITableCache cache, final String oldName, final String newName)
throws Exception {
+ Assert.assertNull(cache.getTable(TABLE_CACHE_TEST_DATABASE, oldName,
false));
+ Assert.assertTrue(
+ getSpecialStatusTable(cache, TABLE_CACHE_TEST_DATABASE, newName)
+ instanceof NonCommittableTsTable);
+ }
+
+ /** A placeholder for a table to be re-fetched only shows up in the private
specialStatusMap. */
+ @SuppressWarnings("unchecked")
+ private static TsTable getSpecialStatusTable(
+ final ITableCache cache, final String database, final String tableName)
throws Exception {
+ final Field field =
DataNodeTableCache.class.getDeclaredField("specialStatusMap");
+ field.setAccessible(true);
+ final Map<String, Map<String, Pair<TsTable, Long>>> specialStatusMap =
+ (Map<String, Map<String, Pair<TsTable, Long>>>) field.get(cache);
+ final Map<String, Pair<TsTable, Long>> tableMap =
+ specialStatusMap.get(PathUtils.unQualifyDatabaseName(database));
+ if (Objects.isNull(tableMap)) {
+ return null;
+ }
+ final Pair<TsTable, Long> tablePair = tableMap.get(tableName);
+ return Objects.isNull(tablePair) ? null : tablePair.getLeft();
+ }
+
private Semaphore getFetchTableSemaphore(final ITableCache cache) throws
Exception {
final Field field =
DataNodeTableCache.class.getDeclaredField("fetchTableSemaphore");
field.setAccessible(true);
