From c4ef5cb5f9dd0e7b801ffd40bfe955f3eca34677 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 11 Sep 2026 18:10:16 +0800 Subject: [PATCH 1/2] Show pre-deleted table schemas and improve write errors --- .../manager/schema/ClusterSchemaManager.java | 3 +- .../persistence/schema/ClusterSchemaInfo.java | 38 ++- .../persistence/schema/ConfigMTree.java | 31 +- .../schema/table/CreateTableProcedure.java | 16 +- .../persistence/schema/ConfigMTreeTest.java | 4 +- .../schema/TablePreDeleteTest.java | 299 ++++++++++++++++++ .../table/CreateTableProcedureTest.java | 39 +++ .../iotdb/db/i18n/DataNodeSchemaMessages.java | 2 - .../iotdb/db/i18n/DataNodeSchemaMessages.java | 2 - .../executor/ClusterConfigTaskExecutor.java | 14 + .../fetcher/TableHeaderSchemaValidator.java | 35 +- .../table/DataNodeTableCache.java | 7 +- .../TableHeaderSchemaValidatorTest.java | 226 +++++++++++++ .../table/DataNodeTableCacheTest.java | 62 ++++ .../iotdb/commons/i18n/CommonMessages.java | 4 + .../iotdb/commons/i18n/CommonMessages.java | 4 + .../table/ColumnInDeletionException.java | 40 +++ .../table/TableInDeletionException.java | 38 +++ 18 files changed, 836 insertions(+), 28 deletions(-) create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidatorTest.java create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInDeletionException.java create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/TableInDeletionException.java diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java index 840be42d48c1..98d9d59088b6 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java @@ -1546,7 +1546,8 @@ public synchronized Pair tableColumnCheckForColumnExtension( final List columnSchemaList, final boolean isTableView) throws MetadataException { - final TsTable originalTable = getTableIfExists(database, tableName).orElse(null); + final TsTable originalTable = + clusterSchemaInfo.getTableForColumnExtension(database, tableName, columnSchemaList); if (Objects.isNull(originalTable)) { return new Pair<>( diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java index 1b9c880cd6a0..8ca27271313f 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java @@ -27,6 +27,8 @@ import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.exception.table.ColumnInDeletionException; +import org.apache.iotdb.commons.exception.table.TableInDeletionException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.path.PathPatternTree; import org.apache.iotdb.commons.schema.table.TableNodeStatus; @@ -34,6 +36,7 @@ import org.apache.iotdb.commons.schema.table.TreeViewSchema; import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema; import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.commons.snapshot.SnapshotProcessor; import org.apache.iotdb.commons.utils.PathUtils; @@ -1352,9 +1355,11 @@ public ShowTableResp showTables(final ShowTablePlan plan) { }) .collect(Collectors.toList()) : tableModelMTree - .getAllUsingTablesUnderSpecificDatabase( + .getAllTablesUnderSpecificDatabase( getQualifiedDatabasePartialPath(plan.getDatabase())) .stream() + .filter(pair -> pair.getRight() != TableNodeStatus.PRE_CREATE) + .map(Pair::getLeft) .map( tsTable -> new TTableInfo( @@ -1447,7 +1452,7 @@ public DescTableResp descTable(final DescTablePlan plan) { } return new DescTableResp( StatusUtils.OK, - tableModelMTree.getUsingTableSchema(databasePath, plan.getTableName()), + tableModelMTree.getTableSchemaForDesc(databasePath, plan.getTableName()), null, null); } catch (final MetadataException e) { @@ -1557,6 +1562,35 @@ public Optional> getTsTableIfExists( } } + public TsTable getTableForColumnExtension( + final String database, + final String tableName, + final List columnSchemaList) + throws MetadataException { + databaseReadWriteLock.readLock().lock(); + try { + final PartialPath databasePath = getQualifiedDatabasePartialPath(database); + final Optional> tableAndStatus = + tableModelMTree.getTableAndStatusIfExists(databasePath, tableName); + if (!tableAndStatus.isPresent()) { + return null; + } + if (tableAndStatus.get().getRight() == TableNodeStatus.PRE_DELETE) { + throw new TableInDeletionException(database, tableName); + } + final TableSchemaDetails details = + tableModelMTree.getTableSchemaDetails(databasePath, tableName); + for (final TsTableColumnSchema column : columnSchemaList) { + if (details.preDeletedColumns.contains(column.getColumnName())) { + throw new ColumnInDeletionException(database, tableName, column.getColumnName()); + } + } + return tableModelMTree.getTableSchemaForDataNode(databasePath, tableName); + } finally { + databaseReadWriteLock.readLock().unlock(); + } + } + public TSStatus addTableColumn(final AddTableColumnPlan plan) { return executeWithLock( () -> { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java index 12f6a2baaa18..8b6b8ca8452c 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java @@ -27,6 +27,7 @@ import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.exception.table.ColumnNotExistsException; import org.apache.iotdb.commons.exception.table.TableAlreadyExistsException; +import org.apache.iotdb.commons.exception.table.TableInDeletionException; import org.apache.iotdb.commons.exception.table.TableNotExistsException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.path.PathPatternTree; @@ -684,6 +685,9 @@ public void preCreateTable(final PartialPath database, final TsTable table) tableNode.setTable(table); tableNode.setStatus(TableNodeStatus.PRE_CREATE); } else if (node instanceof ConfigTableNode) { + if (((ConfigTableNode) node).getStatus() == TableNodeStatus.PRE_DELETE) { + throw new TableInDeletionException(database.getFullPath(), table.getTableName()); + } throw new TableAlreadyExistsException( database.getFullPath().substring(ROOT.length() + 1), table.getTableName()); } else { @@ -839,7 +843,7 @@ public List getAllUsingTablesUnderSpecificDatabase(final PartialPath da child -> child instanceof ConfigTableNode && ((ConfigTableNode) child).getStatus().equals(TableNodeStatus.USING)) - .map(child -> ((ConfigTableNode) child).getTable()) + .map(child -> getTableSchemaForDataNode((ConfigTableNode) child)) .collect(Collectors.toList()); } @@ -869,7 +873,7 @@ public Map getSpecificTablesUnderSpecificDatabase( TsTable table = ((ConfigTableNode) child).getStatus() == TableNodeStatus.PRE_DELETE ? new PreDeleteTsTable(tableName) - : ((ConfigTableNode) child).getTable(); + : getTableSchemaForDataNode((ConfigTableNode) child); result.put(tableName, table); } else { result.put(tableName, null); @@ -1062,16 +1066,29 @@ public void commitAlterColumnDataType( } } - public TsTable getUsingTableSchema(final PartialPath database, final String tableName) + public TsTable getTableSchemaForDataNode(final PartialPath database, final String tableName) + throws MetadataException { + return getTableSchemaForDataNode(getTableNode(database, tableName)); + } + + private TsTable getTableSchemaForDataNode(final ConfigTableNode node) { + if (node.getPreDeletedColumns().isEmpty()) { + return node.getTable(); + } + // Cache reloads and later schema updates must not make a column writable again while its + // deletion is still pending. DESC uses the complete schema separately. + final TsTable table = new TsTable(node.getTable()); + node.getPreDeletedColumns().forEach(table::removeColumnSchema); + return table; + } + + public TsTable getTableSchemaForDesc(final PartialPath database, final String tableName) throws MetadataException { final ConfigTableNode node = getTableNode(database, tableName); - if (node.getPreDeletedColumns().isEmpty() && node.getPreAlteredColumns().isEmpty()) { + if (node.getPreAlteredColumns().isEmpty()) { return node.getTable(); } final TsTable newTable = new TsTable(node.getTable()); - if (!node.getPreDeletedColumns().isEmpty()) { - node.getPreDeletedColumns().forEach(newTable::removeColumnSchema); - } if (!node.getPreAlteredColumns().isEmpty()) { node.getPreAlteredColumns() .forEach( 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 f2379ae1256e..8c6596cfe9ec 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 @@ -22,6 +22,8 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.exception.IoTDBException; import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.exception.table.TableInDeletionException; +import org.apache.iotdb.commons.schema.table.TableNodeStatus; import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan; import org.apache.iotdb.confignode.consensus.request.write.table.CommitCreateTablePlan; @@ -40,6 +42,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.tsfile.utils.Pair; import org.apache.tsfile.utils.ReadWriteIOUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -49,6 +52,7 @@ import java.nio.ByteBuffer; import java.util.Map; import java.util.Objects; +import java.util.Optional; import static org.apache.iotdb.rpc.TSStatusCode.TABLE_ALREADY_EXISTS; @@ -117,10 +121,14 @@ protected Flow executeFromState(final ConfigNodeProcedureEnv env, final CreateTa protected void checkTableExistence(final ConfigNodeProcedureEnv env) { try { - if (env.getConfigManager() - .getClusterSchemaManager() - .getTableIfExists(database, table.getTableName()) - .isPresent()) { + final Optional> existingTable = + env.getConfigManager() + .getClusterSchemaManager() + .getTableAndStatusIfExists(database, table.getTableName()); + if (existingTable.isPresent()) { + if (existingTable.get().getRight() == TableNodeStatus.PRE_DELETE) { + throw new TableInDeletionException(database, table.getTableName()); + } setFailure( new ProcedureException( new IoTDBException( diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java index c2519dcbfae1..2ad27efd1153 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java @@ -434,7 +434,7 @@ public void testAlterColumnTypeUpdatesCompatibleEncoding() throws Exception { SchemaUtils.getDataTypeCompatibleEncoding(TSDataType.STRING, TSEncoding.GORILLA); Assert.assertNotEquals(TSEncoding.GORILLA, expectedEncoding); - final TsTable preAlteredTable = root.getUsingTableSchema(database, table.getTableName()); + final TsTable preAlteredTable = root.getTableSchemaForDesc(database, table.getTableName()); final FieldColumnSchema preAlteredField = (FieldColumnSchema) preAlteredTable.getColumnSchema("measurement"); Assert.assertEquals(TSDataType.STRING, preAlteredField.getDataType()); @@ -443,7 +443,7 @@ public void testAlterColumnTypeUpdatesCompatibleEncoding() throws Exception { root.commitAlterColumnDataType( database, table.getTableName(), "measurement", TSDataType.STRING); - final TsTable committedTable = root.getUsingTableSchema(database, table.getTableName()); + final TsTable committedTable = root.getTableSchemaForDesc(database, table.getTableName()); final FieldColumnSchema committedField = (FieldColumnSchema) committedTable.getColumnSchema("measurement"); Assert.assertEquals(TSDataType.STRING, committedField.getDataType()); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java new file mode 100644 index 000000000000..10984893d60a --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java @@ -0,0 +1,299 @@ +/* + * 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.persistence.schema; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.exception.table.ColumnInDeletionException; +import org.apache.iotdb.commons.exception.table.TableInDeletionException; +import org.apache.iotdb.commons.schema.table.TableNodeStatus; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil; +import org.apache.iotdb.commons.schema.table.column.AttributeColumnSchema; +import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema; +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType; +import org.apache.iotdb.confignode.consensus.request.read.table.DescTablePlan; +import org.apache.iotdb.confignode.consensus.request.read.table.FetchTablePlan; +import org.apache.iotdb.confignode.consensus.request.read.table.ShowTablePlan; +import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan; +import org.apache.iotdb.confignode.consensus.request.write.table.CommitCreateTablePlan; +import org.apache.iotdb.confignode.consensus.request.write.table.CommitDeleteColumnPlan; +import org.apache.iotdb.confignode.consensus.request.write.table.CommitDeleteTablePlan; +import org.apache.iotdb.confignode.consensus.request.write.table.PreAlterColumnDataTypePlan; +import org.apache.iotdb.confignode.consensus.request.write.table.PreCreateTablePlan; +import org.apache.iotdb.confignode.consensus.request.write.table.PreDeleteColumnPlan; +import org.apache.iotdb.confignode.consensus.request.write.table.PreDeleteTablePlan; +import org.apache.iotdb.confignode.manager.IManager; +import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager; +import org.apache.iotdb.confignode.manager.schema.ClusterSchemaQuotaStatistics; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; +import org.apache.iotdb.confignode.rpc.thrift.TDescTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TTableInfo; +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.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +public class TablePreDeleteTest { + private static final String DATABASE = "root.pre_delete_test"; + private static final String TABLE = "table1"; + private ClusterSchemaInfo schemaInfo; + private ClusterSchemaManager schemaManager; + + @Before + public void setUp() throws Exception { + schemaInfo = new ClusterSchemaInfo(); + schemaManager = + new ClusterSchemaManager( + Mockito.mock(IManager.class), + schemaInfo, + Mockito.mock(ClusterSchemaQuotaStatistics.class)); + assertSuccess( + schemaInfo.createDatabase( + new DatabaseSchemaPlan( + ConfigPhysicalPlanType.CreateDatabase, + new TDatabaseSchema(DATABASE).setIsTableModel(true)))); + createTable(TABLE); + } + + @After + public void tearDown() { + schemaInfo.clear(); + } + + @Test + public void testShowTablesIncludesPreDeleteButNotPreCreate() { + createTable("using_table"); + assertSuccess( + schemaInfo.preCreateTable(new PreCreateTablePlan(DATABASE, new TsTable("creating")))); + assertSuccess(schemaInfo.preDeleteTable(new PreDeleteTablePlan(DATABASE, TABLE))); + + final TShowTableResp basic = + schemaInfo.showTables(new ShowTablePlan(DATABASE, false)).convertToTShowTableResp(); + assertSuccess(basic.getStatus()); + assertEquals( + Arrays.asList(TABLE, "using_table"), + basic.getTableInfoList().stream() + .map(TTableInfo::getTableName) + .sorted() + .collect(Collectors.toList())); + + final TShowTableResp details = + schemaInfo.showTables(new ShowTablePlan(DATABASE, true)).convertToTShowTableResp(); + final Map states = + details.getTableInfoList().stream() + .collect(Collectors.toMap(TTableInfo::getTableName, TTableInfo::getState)); + assertEquals(Integer.valueOf(TableNodeStatus.PRE_DELETE.ordinal()), states.get(TABLE)); + assertEquals(Integer.valueOf(TableNodeStatus.PRE_CREATE.ordinal()), states.get("creating")); + assertEquals(Integer.valueOf(TableNodeStatus.USING.ordinal()), states.get("using_table")); + + assertSuccess(schemaInfo.dropTable(new CommitDeleteTablePlan(DATABASE, TABLE))); + assertEquals( + Collections.singletonList("using_table"), + schemaInfo + .showTables(new ShowTablePlan(DATABASE, false)) + .convertToTShowTableResp() + .getTableInfoList() + .stream() + .map(TTableInfo::getTableName) + .collect(Collectors.toList())); + } + + @Test + public void testDescRetainsPreDeletedColumnsAndAlteredTypes() { + assertSuccess(schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE, TABLE, "field"))); + assertSuccess( + schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE, TABLE, "attribute"))); + assertSuccess( + schemaInfo.preAlterColumnDataType( + new PreAlterColumnDataTypePlan(DATABASE, TABLE, "live", TSDataType.INT64))); + + TDescTableResp basic = describe(false); + TsTable table = TsTableInternalRPCUtil.deserializeSingleTsTable(basic.getTableInfo()); + assertNotNull(table.getColumnSchema("field")); + assertNotNull(table.getColumnSchema("attribute")); + assertEquals(TSDataType.INT64, table.getColumnSchema("live").getDataType()); + assertFalse(basic.isSetPreDeletedColumns()); + + final TDescTableResp details = describe(true); + assertTrue(details.getPreDeletedColumns().containsAll(Arrays.asList("field", "attribute"))); + assertEquals( + Byte.valueOf(TSDataType.INT64.serialize()), details.getPreAlteredColumns().get("live")); + + assertSuccess( + schemaInfo.commitDeleteColumn(new CommitDeleteColumnPlan(DATABASE, TABLE, "field"))); + table = TsTableInternalRPCUtil.deserializeSingleTsTable(describe(false).getTableInfo()); + assertNull(table.getColumnSchema("field")); + + assertSuccess(schemaInfo.preDeleteTable(new PreDeleteTablePlan(DATABASE, TABLE))); + assertNotNull( + TsTableInternalRPCUtil.deserializeSingleTsTable(describe(false).getTableInfo()) + .getColumnSchema("attribute")); + } + + @Test + public void testColumnExtensionRejectsPreDeletedNames() throws Exception { + for (final String column : Arrays.asList("field", "attribute")) { + assertSuccess(schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE, TABLE, column))); + final List columns = + new ArrayList<>(Arrays.asList(field("new_field"), field(column))); + final ColumnInDeletionException exception = + assertThrows( + ColumnInDeletionException.class, + () -> + schemaManager.tableColumnCheckForColumnExtension( + DATABASE, TABLE, columns, false)); + assertEquals(TSStatusCode.SEMANTIC_ERROR.getStatusCode(), exception.getErrorCode()); + assertEquals( + new ColumnInDeletionException(DATABASE, TABLE, column).getMessage(), + exception.getMessage()); + assertEquals(2, columns.size()); + assertNull( + schemaInfo + .getTsTableIfExists(DATABASE, TABLE) + .get() + .getLeft() + .getColumnSchema("new_field")); + } + assertEquals( + TSStatusCode.COLUMN_ALREADY_EXISTS.getStatusCode(), + schemaManager + .tableColumnCheckForColumnExtension( + DATABASE, TABLE, new ArrayList<>(Collections.singletonList(field("live"))), false) + .getLeft() + .getCode()); + + assertSuccess( + schemaInfo.commitDeleteColumn(new CommitDeleteColumnPlan(DATABASE, TABLE, "field"))); + assertSuccess( + schemaManager + .tableColumnCheckForColumnExtension( + DATABASE, TABLE, new ArrayList<>(Collections.singletonList(field("field"))), false) + .getLeft()); + } + + @Test + public void testPreDeletedTableRejectsCreationAndColumnExtension() { + assertSuccess(schemaInfo.preDeleteTable(new PreDeleteTablePlan(DATABASE, TABLE))); + final TSStatus create = + schemaInfo.preCreateTable(new PreCreateTablePlan(DATABASE, new TsTable(TABLE))); + assertEquals(TSStatusCode.SEMANTIC_ERROR.getStatusCode(), create.getCode()); + assertEquals(new TableInDeletionException(DATABASE, TABLE).getMessage(), create.getMessage()); + final TableInDeletionException exception = + assertThrows( + TableInDeletionException.class, + () -> + schemaManager.tableColumnCheckForColumnExtension( + DATABASE, + TABLE, + new ArrayList<>(Collections.singletonList(field("new_field"))), + false)); + assertEquals(create.getMessage(), exception.getMessage()); + } + + @Test + public void testDataNodeSchemasExcludePreDeletedColumns() throws Exception { + assertSuccess(schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE, TABLE, "field"))); + assertSuccess( + schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE, TABLE, "attribute"))); + + assertOnlyLiveColumns(schemaInfo.getAllUsingTables().get(DATABASE).get(0)); + final TsTable fetched = + TsTableInternalRPCUtil.deserializeTsTableFetchResult( + schemaInfo + .fetchTables( + new FetchTablePlan( + Collections.singletonMap(DATABASE, Collections.singleton(TABLE)), + Collections.singleton(TableNodeStatus.USING))) + .convertToTFetchTableResp() + .getTableInfoMap()) + .get(DATABASE) + .get(TABLE); + assertOnlyLiveColumns(fetched); + + final TsTable expanded = + schemaManager + .tableColumnCheckForColumnExtension( + DATABASE, + TABLE, + new ArrayList<>(Collections.singletonList(field("new_field"))), + false) + .getRight(); + assertOnlyLiveColumns(expanded); + assertNotNull(expanded.getColumnSchema("new_field")); + + // Preparing a cache snapshot must not change the complete schema used by DESC. + final TsTable described = + TsTableInternalRPCUtil.deserializeSingleTsTable(describe(false).getTableInfo()); + assertNotNull(described.getColumnSchema("field")); + assertNotNull(described.getColumnSchema("attribute")); + } + + private static void assertOnlyLiveColumns(final TsTable table) { + assertNotNull(table.getColumnSchema("live")); + assertNull(table.getColumnSchema("field")); + assertNull(table.getColumnSchema("attribute")); + } + + private TDescTableResp describe(final boolean details) { + final TDescTableResp resp = + schemaInfo.descTable(new DescTablePlan(DATABASE, TABLE, details)).convertToTDescTableResp(); + assertSuccess(resp.getStatus()); + return resp; + } + + private void createTable(final String name) { + final TsTable table = new TsTable(name); + table.addColumnSchema(field("field")); + table.addColumnSchema(field("live")); + table.addColumnSchema(new AttributeColumnSchema("attribute", TSDataType.STRING)); + assertSuccess(schemaInfo.preCreateTable(new PreCreateTablePlan(DATABASE, table))); + assertSuccess(schemaInfo.commitCreateTable(new CommitCreateTablePlan(DATABASE, name))); + } + + private static FieldColumnSchema field(final String name) { + return new FieldColumnSchema(name, TSDataType.INT32, TSEncoding.RLE, CompressionType.LZ4); + } + + private static void assertSuccess(final TSStatus status) { + assertEquals( + status.getMessage(), TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedureTest.java index fab1bfb26601..0b9a118fd1a2 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedureTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedureTest.java @@ -20,24 +20,63 @@ package org.apache.iotdb.confignode.procedure.impl.schema.table; import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.exception.IoTDBException; +import org.apache.iotdb.commons.exception.table.TableInDeletionException; +import org.apache.iotdb.commons.schema.table.TableNodeStatus; import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.commons.schema.table.column.AttributeColumnSchema; import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.confignode.procedure.store.ProcedureType; +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; +import org.mockito.Mockito; import java.io.ByteArrayOutputStream; import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Optional; public class CreateTableProcedureTest { + @Test + public void testPreDeletedTableIsNotReportedAsAlreadyExisting() throws Exception { + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final ClusterSchemaManager schemaManager = Mockito.mock(ClusterSchemaManager.class); + Mockito.when(env.getConfigManager()).thenReturn(configManager); + Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager); + final TsTable table = new TsTable("table1"); + Mockito.when(schemaManager.getTableAndStatusIfExists("database1", "table1")) + .thenReturn(Optional.of(new Pair<>(table, TableNodeStatus.PRE_DELETE))); + + final CreateTableProcedure procedure = new CreateTableProcedure("database1", table, false); + procedure.checkTableExistence(env); + + Assert.assertTrue(procedure.isFailed()); + Assert.assertTrue(procedure.getException().getCause() instanceof TableInDeletionException); + Assert.assertEquals( + TSStatusCode.SEMANTIC_ERROR.getStatusCode(), + ((IoTDBException) procedure.getException().getCause()).getErrorCode()); + + Mockito.when(schemaManager.getTableAndStatusIfExists("database1", "table1")) + .thenReturn(Optional.of(new Pair<>(table, TableNodeStatus.USING))); + final CreateTableProcedure duplicate = new CreateTableProcedure("database1", table, false); + duplicate.checkTableExistence(env); + Assert.assertEquals( + TSStatusCode.TABLE_ALREADY_EXISTS.getStatusCode(), + ((IoTDBException) duplicate.getException().getCause()).getErrorCode()); + } + @Test public void serializeDeserializeTest() throws IllegalPathException, IOException { final TsTable table = new TsTable("table1"); diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java index dd1c4af94f64..ef703af3f0e3 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java @@ -584,8 +584,6 @@ public final class DataNodeSchemaMessages { public static final String UPDATE_TABLE_BY_FETCH_WITH_DETAIL = "Update table {}.{} by table fetch, {}"; public static final String UPDATE_TABLE_BY_FETCH = "Update table {}.{} by table fetch."; - public static final String THE_TABLE_IS_IN_PRE_DELETE_STATE = - "The table %s.%s is in the pre-delete state. Please wait a few seconds. If the table is still in this state, please drop it again."; public static final String COMPARE_TABLE_ADDED = "Added table: "; public static final String COMPARE_TABLE_REMOVED = "Removed table: "; public static final String COMPARE_TABLE_NAME = "Table name: "; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java index 411044b14afb..f1ceaaa0470d 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java @@ -578,8 +578,6 @@ public final class DataNodeSchemaMessages { "尝试获取信号量以从 ConfigNode 获取表时被中断，已忽略。"; public static final String UPDATE_TABLE_BY_FETCH_WITH_DETAIL = "获取表 {}.{} 信息, {}"; public static final String UPDATE_TABLE_BY_FETCH = "通过表拉取更新表 {}.{}"; - public static final String THE_TABLE_IS_IN_PRE_DELETE_STATE = - "表 %s.%s 处于预删除的状态，请稍等，如之后重试还是此状态，请输入sql再次删除"; public static final String COMPARE_TABLE_ADDED = "新增表："; public static final String COMPARE_TABLE_REMOVED = "已移除表："; public static final String COMPARE_TABLE_NAME = "表名："; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java index 3ddd2531200d..18152effaecc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java @@ -4819,6 +4819,20 @@ public SettableFuture createTable( return future; } + public Set getPreDeletedColumns(final String database, final String tableName) { + try (final ConfigNodeClient configNodeClient = + CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { + final TDescTableResp resp = configNodeClient.describeTable(database, tableName, true); + if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new IoTDBRuntimeException( + getTableErrorMessage(resp.getStatus(), database), resp.getStatus().getCode()); + } + return resp.isSetPreDeletedColumns() ? resp.getPreDeletedColumns() : Collections.emptySet(); + } catch (final ClientManagerException | TException e) { + throw new RuntimeException(e); + } + } + @Override public SettableFuture describeTable( final String database, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java index 6a08427d82b2..8098d382a465 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.exception.table.ColumnInDeletionException; import org.apache.iotdb.commons.i18n.QueryMessages; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; @@ -71,6 +72,7 @@ import java.util.List; import java.util.Objects; import java.util.Optional; +import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; @@ -83,11 +85,14 @@ public class TableHeaderSchemaValidator { private static final Logger LOGGER = LoggerFactory.getLogger(TableHeaderSchemaValidator.class); - private final ClusterConfigTaskExecutor configTaskExecutor = - ClusterConfigTaskExecutor.getInstance(); + private final ClusterConfigTaskExecutor configTaskExecutor; private TableHeaderSchemaValidator() { - // do nothing + this(ClusterConfigTaskExecutor.getInstance()); + } + + TableHeaderSchemaValidator(final ClusterConfigTaskExecutor configTaskExecutor) { + this.configTaskExecutor = configTaskExecutor; } private static class TableHeaderSchemaValidatorHolder { @@ -213,6 +218,7 @@ public Optional validateTableHeaderSchema4TsFile( boolean refreshed = false; boolean noField = true; + Set preDeletedColumns = null; for (final ColumnSchema columnSchema : inputColumnList) { TsTableColumnSchema existingColumn = table.getColumnSchema(columnSchema.getName()); if (Objects.isNull(existingColumn)) { @@ -225,6 +231,12 @@ public Optional validateTableHeaderSchema4TsFile( existingColumn = table.getColumnSchema(columnSchema.getName()); } if (Objects.isNull(existingColumn)) { + if (preDeletedColumns == null) { + preDeletedColumns = + configTaskExecutor.getPreDeletedColumns(database, tableSchema.getTableName()); + } + checkColumnNotPreDeleted( + database, tableSchema.getTableName(), columnSchema.getName(), preDeletedColumns); // check arguments for column auto creation if (columnSchema.getColumnCategory() == null) { throw new SemanticException( @@ -397,6 +409,7 @@ public void validateInsertNodeMeasurements( boolean refreshed = false; boolean noField = true; boolean hasAttribute = false; + Set preDeletedColumns = null; // Track TAG column measurement indices for batch processing after validation loop // LinkedHashMap maintains insertion order, key is column name, value is measurement index @@ -432,6 +445,12 @@ public void validateInsertNodeMeasurements( } if (Objects.isNull(existingColumn)) { + if (preDeletedColumns == null) { + preDeletedColumns = + configTaskExecutor.getPreDeletedColumns(database, measurementInfo.getTableName()); + } + checkColumnNotPreDeleted( + database, measurementInfo.getTableName(), measurementName, preDeletedColumns); // Check arguments for column auto creation if (category == null) { throw new SemanticException( @@ -539,6 +558,16 @@ public void validateInsertNodeMeasurements( } } + private static void checkColumnNotPreDeleted( + final String database, + final String tableName, + final String columnName, + final Set preDeletedColumns) { + if (preDeletedColumns.contains(columnName)) { + throw new SemanticException(new ColumnInDeletionException(database, tableName, columnName)); + } + } + private void autoCreateTableFromMeasurementInfo( final MPPQueryContext context, final String database, 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 23daf268206c..88b09d190747 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 @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy; import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.exception.table.TableInDeletionException; import org.apache.iotdb.commons.schema.table.NonCommittableTsTable; import org.apache.iotdb.commons.schema.table.PreDeleteTsTable; import org.apache.iotdb.commons.schema.table.TableNodeStatus; @@ -708,11 +709,7 @@ private void updateDeleteTable( instanceVersion.incrementAndGet(); } if (targetTableIsStillDeleting) { - throw new SemanticException( - String.format( - DataNodeSchemaMessages.THE_TABLE_IS_IN_PRE_DELETE_STATE, - targetDatabase, - targetTable)); + throw new SemanticException(new TableInDeletionException(targetDatabase, targetTable)); } } finally { readWriteLock.writeLock().unlock(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidatorTest.java new file mode 100644 index 000000000000..2588031d9861 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidatorTest.java @@ -0,0 +1,226 @@ +/* + * 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.db.queryengine.plan.relational.metadata.fetcher; + +import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.exception.table.ColumnInDeletionException; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.TableSchema; +import org.apache.iotdb.commons.schema.table.InsertNodeMeasurementInfo; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.AttributeColumnSchema; +import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.conf.IoTDBConfig; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; +import org.apache.iotdb.db.queryengine.plan.execution.config.executor.ClusterConfigTaskExecutor; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.schemaengine.table.ITableCache; +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.read.common.type.TypeFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +public class TableHeaderSchemaValidatorTest { + private static final String DATABASE = "pre_delete_write_test"; + private static final String TABLE = "table1"; + private final ITableCache cache = DataNodeTableCache.getInstance(); + private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + private final MPPQueryContext context = new MPPQueryContext(new QueryId("pre_delete_write_test")); + private ClusterConfigTaskExecutor executor; + private TableHeaderSchemaValidator validator; + private boolean autoCreateSchema; + private boolean partialInsert; + + @Before + public void setUp() { + autoCreateSchema = config.isAutoCreateSchemaEnabled(); + partialInsert = config.isEnablePartialInsert(); + executor = Mockito.mock(ClusterConfigTaskExecutor.class); + validator = new TableHeaderSchemaValidator(executor); + cache.invalid(DATABASE); + final TsTable table = new TsTable(TABLE); + table.addColumnSchema( + new FieldColumnSchema("live", TSDataType.INT32, TSEncoding.RLE, CompressionType.LZ4)); + table.addColumnSchema( + new FieldColumnSchema("field", TSDataType.INT32, TSEncoding.RLE, CompressionType.LZ4)); + table.addColumnSchema(new AttributeColumnSchema("attribute", TSDataType.STRING)); + cache.preUpdateTable(DATABASE, table, null); + cache.commitUpdateTable(DATABASE, TABLE, null); + // A DROP COLUMN invalidates the DataNode cache before removing ConfigNode metadata. + cache.invalid(DATABASE, TABLE, "field"); + cache.invalid(DATABASE, TABLE, "attribute"); + Mockito.when(executor.getPreDeletedColumns(DATABASE, TABLE)) + .thenReturn(new java.util.HashSet<>(Arrays.asList("field", "attribute"))); + } + + @After + public void tearDown() { + DataNodeSchemaLockManager.getInstance().releaseReadLock(context); + cache.invalid(DATABASE); + config.setAutoCreateSchemaEnabled(autoCreateSchema); + config.setEnablePartialInsert(partialInsert); + } + + @Test + public void testInsertReportsDeletionBeforeUnknownCategory() { + for (final boolean autoCreate : new boolean[] {true, false}) { + config.setAutoCreateSchemaEnabled(autoCreate); + final InsertNodeMeasurementInfo measurements = measurements("field", null); + final SemanticException error = + assertThrows( + SemanticException.class, + () -> + validator.validateInsertNodeMeasurements( + DATABASE, measurements, context, true, null, null)); + assertDeletion(error, "field"); + } + } + + @Test + public void testInsertRejectsPreDeletedAttribute() { + final SemanticException error = + assertThrows( + SemanticException.class, + () -> + validator.validateInsertNodeMeasurements( + DATABASE, + measurements("attribute", TsTableColumnCategory.ATTRIBUTE), + context, + true, + null, + null)); + assertDeletion(error, "attribute"); + } + + @Test + public void testTsFileLoadRejectsPreDeletedFieldAndAttribute() { + for (final boolean autoCreate : new boolean[] {true, false}) { + config.setAutoCreateSchemaEnabled(autoCreate); + for (final String column : Arrays.asList("field", "attribute")) { + final TableSchema tableSchema = + new TableSchema( + TABLE, + Collections.singletonList( + new ColumnSchema( + column, + TypeFactory.getType( + column.equals("field") ? TSDataType.INT32 : TSDataType.STRING), + false, + column.equals("field") + ? TsTableColumnCategory.FIELD + : TsTableColumnCategory.ATTRIBUTE))); + final SemanticException error = + assertThrows( + SemanticException.class, + () -> + validator.validateTableHeaderSchema4TsFile( + DATABASE, tableSchema, context, true, false, new AtomicBoolean())); + assertDeletion(error, column); + } + } + } + + @Test + public void testExistingColumnsDoNotFetchDeletionStatus() throws Exception { + validator.validateInsertNodeMeasurements( + DATABASE, measurements("live", TsTableColumnCategory.FIELD), context, true, null, null); + validator.validateTableHeaderSchema4TsFile( + DATABASE, + new TableSchema( + TABLE, + Collections.singletonList( + new ColumnSchema( + "live", + TypeFactory.getType(TSDataType.INT32), + false, + TsTableColumnCategory.FIELD))), + context, + true, + false, + new AtomicBoolean()); + Mockito.verify(executor, Mockito.never()) + .getPreDeletedColumns(Mockito.anyString(), Mockito.anyString()); + } + + @Test + public void testMissingColumnStillReportsUnknownCategory() { + final SemanticException error = + assertThrows( + SemanticException.class, + () -> + validator.validateInsertNodeMeasurements( + DATABASE, measurements("missing", null), context, true, null, null)); + assertEquals(TSStatusCode.COLUMN_NOT_EXISTS.getStatusCode(), error.getErrorCode()); + } + + @Test + public void testMissingColumnsFetchDeletionStatusOnce() { + config.setAutoCreateSchemaEnabled(false); + config.setEnablePartialInsert(true); + final InsertNodeMeasurementInfo measurements = Mockito.mock(InsertNodeMeasurementInfo.class); + Mockito.when(measurements.getTableName()).thenReturn(TABLE); + Mockito.when(measurements.getMeasurementCount()).thenReturn(2); + Mockito.when(measurements.getColumnCategories()) + .thenReturn( + new TsTableColumnCategory[] {TsTableColumnCategory.FIELD, TsTableColumnCategory.FIELD}); + Mockito.when(measurements.getMeasurementName(0)).thenReturn("missing1"); + Mockito.when(measurements.getMeasurementName(1)).thenReturn("missing2"); + validator.validateInsertNodeMeasurements(DATABASE, measurements, context, true, null, null); + Mockito.verify(executor).getPreDeletedColumns(DATABASE, TABLE); + } + + private InsertNodeMeasurementInfo measurements( + final String name, final TsTableColumnCategory category) { + final InsertNodeMeasurementInfo measurements = Mockito.mock(InsertNodeMeasurementInfo.class); + Mockito.when(measurements.getTableName()).thenReturn(TABLE); + Mockito.when(measurements.getMeasurementCount()).thenReturn(1); + Mockito.when(measurements.getColumnCategories()) + .thenReturn(new TsTableColumnCategory[] {category}); + Mockito.when(measurements.getMeasurementName(0)).thenReturn(name); + Mockito.when(measurements.getType(0)).thenReturn(TSDataType.INT32); + return measurements; + } + + private static void assertDeletion(final SemanticException error, final String column) { + assertEquals(TSStatusCode.SEMANTIC_ERROR.getStatusCode(), error.getErrorCode()); + assertTrue(error.getCause() instanceof ColumnInDeletionException); + assertEquals( + new ColumnInDeletionException(DATABASE, TABLE, column).getMessage(), + error.getCause().getMessage()); + } +} 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 fd97da843bdc..217532f6c99e 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 @@ -19,6 +19,10 @@ 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.exception.table.TableInDeletionException; +import org.apache.iotdb.commons.schema.table.PreDeleteTsTable; import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; @@ -29,6 +33,10 @@ import org.junit.Test; import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.Collections; +import java.util.Map; import java.util.concurrent.Semaphore; public class DataNodeTableCacheTest { @@ -117,6 +125,60 @@ private Semaphore getFetchTableSemaphore(final ITableCache cache) throws Excepti return (Semaphore) field.get(cache); } + @Test + public void preDeletedTableRefreshReportsDeletionAndRecovers() throws Exception { + final ITableCache cache = DataNodeTableCache.getInstance(); + final Method updateDeleteTable = + DataNodeTableCache.class.getDeclaredMethod( + "updateDeleteTable", + Map.class, + String.class, + String.class, + LeaseFencedRetryPolicy.class); + updateDeleteTable.setAccessible(true); + final String database = "pre_delete_table_test"; + cache.invalid(database); + try { + cache.preUpdateTable(database, new PreDeleteTsTable(TABLE_NAME), null); + final InvocationTargetException failure = + Assert.assertThrows( + InvocationTargetException.class, + () -> + updateDeleteTable.invoke( + cache, + Collections.singletonMap( + database, + Collections.singletonMap(TABLE_NAME, new PreDeleteTsTable(TABLE_NAME))), + database, + TABLE_NAME, + LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS)); + Assert.assertTrue(failure.getCause() instanceof SemanticException); + Assert.assertEquals( + new TableInDeletionException(database, TABLE_NAME).getMessage(), + failure.getCause().getCause().getMessage()); + + updateDeleteTable.invoke( + cache, + Collections.singletonMap( + database, Collections.singletonMap(TABLE_NAME, createTable(TABLE_NAME))), + database, + TABLE_NAME, + LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS); + Assert.assertNotNull(cache.getTableInWrite(database, TABLE_NAME)); + + cache.preUpdateTable(database, new PreDeleteTsTable(TABLE_NAME), null); + updateDeleteTable.invoke( + cache, + Collections.singletonMap(database, Collections.singletonMap(TABLE_NAME, null)), + database, + TABLE_NAME, + LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS); + Assert.assertNull(cache.getTableInWrite(database, TABLE_NAME)); + } finally { + cache.invalid(database); + } + } + private TsTable createTable(final String tableName) { final TsTable table = new TsTable(tableName); table.addColumnSchema( diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java index a05c8f9915a1..b7bf58fd8bd3 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -335,4 +335,8 @@ private CommonMessages() {} public static final String EXCEPTION_SNAPSHOT_BUFFER_SIZE_MUST_NOT_EXCEED_ARG_BYTES_BUT_WAS_ARG_D1DA6F7E = "Snapshot buffer size must not exceed %d bytes, but was %d."; + public static final String EXCEPTION_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROP_TABLE_IF_IT_IS_STUCK_7E22D78F = + "Table '%s.%s' is being deleted. Please wait for deletion to finish, or retry DROP TABLE if it is stuck."; + public static final String EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROPPING_THE_COLUMN_IF_IT_IS_STUCK_875DAFFE = + "Column '%s' in table '%s.%s' is being deleted. Please wait for deletion to finish, or retry dropping the column if it is stuck."; } diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index c5c6290687e6..45a682efe376 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -232,4 +232,8 @@ private CommonMessages() {} public static final String EXCEPTION_SNAPSHOT_BUFFER_SIZE_MUST_NOT_EXCEED_ARG_BYTES_BUT_WAS_ARG_D1DA6F7E = "快照缓冲区大小不得超过 %d 字节，但实际为 %d。"; + public static final String EXCEPTION_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROP_TABLE_IF_IT_IS_STUCK_7E22D78F = + "表 '%s.%s' 正在删除中。请等待删除完成；如果删除一直未完成，请重试 DROP TABLE。"; + public static final String EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROPPING_THE_COLUMN_IF_IT_IS_STUCK_875DAFFE = + "列 '%s'（位于表 '%s.%s'）正在删除中。请等待删除完成；如果删除一直未完成，请重试删除该列。"; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInDeletionException.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInDeletionException.java new file mode 100644 index 000000000000..b3a6e60be096 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInDeletionException.java @@ -0,0 +1,40 @@ +/* + * 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.commons.exception.table; + +import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.i18n.CommonMessages; +import org.apache.iotdb.commons.utils.PathUtils; +import org.apache.iotdb.rpc.TSStatusCode; + +public class ColumnInDeletionException extends MetadataException { + + public ColumnInDeletionException( + final String database, final String tableName, final String columnName) { + super( + String.format( + CommonMessages + .EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROPPING_THE_COLUMN_IF_IT_IS_STUCK_875DAFFE, + columnName, + PathUtils.unQualifyDatabaseName(database), + tableName), + TSStatusCode.SEMANTIC_ERROR.getStatusCode()); + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/TableInDeletionException.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/TableInDeletionException.java new file mode 100644 index 000000000000..8ec7d18937cb --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/TableInDeletionException.java @@ -0,0 +1,38 @@ +/* + * 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.commons.exception.table; + +import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.i18n.CommonMessages; +import org.apache.iotdb.commons.utils.PathUtils; +import org.apache.iotdb.rpc.TSStatusCode; + +public class TableInDeletionException extends MetadataException { + + public TableInDeletionException(final String database, final String tableName) { + super( + String.format( + CommonMessages + .EXCEPTION_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROP_TABLE_IF_IT_IS_STUCK_7E22D78F, + PathUtils.unQualifyDatabaseName(database), + tableName), + TSStatusCode.SEMANTIC_ERROR.getStatusCode()); + } +} From 89c80a1792db26b7ac4cf77ffed4ddaa3230438a Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 15 Sep 2026 11:29:47 +0800 Subject: [PATCH 2/2] Support rollback for pre-altered table columns --- .../consensus/request/ConfigPhysicalPlan.java | 4 + .../request/ConfigPhysicalPlanType.java | 1 + .../RollbackPreAlterColumnDataTypePlan.java | 61 +++++++ .../manager/schema/ClusterSchemaManager.java | 39 +++- .../executor/ConfigPlanExecutor.java | 4 + .../persistence/schema/ClusterSchemaInfo.java | 79 +++++++- .../persistence/schema/ConfigMTree.java | 140 ++++++++++++-- .../AlterTableColumnDataTypeProcedure.java | 78 +++++++- .../request/ConfigPhysicalPlanSerDeTest.java | 15 ++ .../persistence/schema/ConfigMTreeTest.java | 73 +++++++- .../schema/TablePreDeleteTest.java | 69 +++++++ ...AlterTableColumnDataTypeProcedureTest.java | 171 ++++++++++++++++++ .../table/DataNodeTableCache.java | 79 +++++++- .../table/DataNodeTableCacheTest.java | 129 ++++++++++++- .../iotdb/commons/i18n/CommonMessages.java | 2 + .../iotdb/commons/i18n/CommonMessages.java | 2 + .../table/ColumnInAlterException.java | 40 ++++ 17 files changed, 956 insertions(+), 30 deletions(-) create mode 100644 iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/RollbackPreAlterColumnDataTypePlan.java create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedureTest.java create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInAlterException.java diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java index 0d9ca912571a..ff765a035b0d 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java @@ -116,6 +116,7 @@ import org.apache.iotdb.confignode.consensus.request.write.table.RenameTableColumnPlan; import org.apache.iotdb.confignode.consensus.request.write.table.RenameTablePlan; import org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan; +import org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreAlterColumnDataTypePlan; import org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreDeleteTablePlan; import org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan; import org.apache.iotdb.confignode.consensus.request.write.table.SetTableCommentPlan; @@ -463,6 +464,9 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept case PreAlterColumnDataType: plan = new PreAlterColumnDataTypePlan(); break; + case RollbackPreAlterColumnDataType: + plan = new RollbackPreAlterColumnDataTypePlan(); + break; case AlterColumnDataType: plan = new AlterColumnDataTypePlan(); break; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java index 1be951814148..afef9f82bfe9 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java @@ -233,6 +233,7 @@ public enum ConfigPhysicalPlanType { AlterColumnDataType((short) 878), PreAlterColumnDataType((short) 879), RollbackPreDeleteTable((short) 880), + RollbackPreAlterColumnDataType((short) 881), /** Deprecated types for sync, restored them for upgrade. */ @Deprecated diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/RollbackPreAlterColumnDataTypePlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/RollbackPreAlterColumnDataTypePlan.java new file mode 100644 index 000000000000..bcb9fb4ca720 --- /dev/null +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/RollbackPreAlterColumnDataTypePlan.java @@ -0,0 +1,61 @@ +/* + * 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.consensus.request.write.table; + +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType; + +import org.apache.tsfile.enums.TSDataType; + +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; + +public class RollbackPreAlterColumnDataTypePlan extends AbstractTableColumnPlan { + private TSDataType newType; + + public RollbackPreAlterColumnDataTypePlan() { + super(ConfigPhysicalPlanType.RollbackPreAlterColumnDataType); + } + + public RollbackPreAlterColumnDataTypePlan( + final String database, + final String tableName, + final String columnName, + final TSDataType newType) { + super(ConfigPhysicalPlanType.RollbackPreAlterColumnDataType, database, tableName, columnName); + this.newType = newType; + } + + @Override + protected void serializeImpl(final DataOutputStream stream) throws IOException { + super.serializeImpl(stream); + stream.write(newType.serialize()); + } + + @Override + protected void deserializeImpl(final ByteBuffer buffer) throws IOException { + super.deserializeImpl(buffer); + newType = TSDataType.deserializeFrom(buffer); + } + + public TSDataType getNewType() { + return newType; + } +} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java index 98d9d59088b6..b55180e35f75 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java @@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.exception.table.TableInDeletionException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.path.PathPatternTree; import org.apache.iotdb.commons.schema.SchemaConstant; @@ -1540,6 +1541,21 @@ public Optional> getTableAndStatusIfExists( return clusterSchemaInfo.getTsTableIfExists(database, tableName); } + public boolean isColumnAlterCommitted( + final String database, + final String tableName, + final String columnName, + final TSDataType dataType) + throws MetadataException { + return clusterSchemaInfo.isColumnAlterCommitted(database, tableName, columnName, dataType); + } + + public Optional getPreAlteredColumnType( + final String database, final String tableName, final String columnName) + throws MetadataException { + return clusterSchemaInfo.getPreAlteredColumnType(database, tableName, columnName); + } + public synchronized Pair tableColumnCheckForColumnExtension( final String database, final String tableName, @@ -1547,7 +1563,12 @@ public synchronized Pair tableColumnCheckForColumnExtension( final boolean isTableView) throws MetadataException { final TsTable originalTable = - clusterSchemaInfo.getTableForColumnExtension(database, tableName, columnSchemaList); + clusterSchemaInfo.getTableForModification( + database, + tableName, + columnSchemaList.stream() + .map(TsTableColumnSchema::getColumnName) + .toArray(String[]::new)); if (Objects.isNull(originalTable)) { return new Pair<>( @@ -1602,7 +1623,7 @@ public synchronized Pair tableColumnCheckForColumnAltering( final TSDataType dataType, final boolean isGeneratedByPipe) throws MetadataException { - final TsTable originalTable = getTableIfExists(database, tableName).orElse(null); + final TsTable originalTable = clusterSchemaInfo.getTableForModification(database, tableName); if (Objects.isNull(originalTable)) { return new Pair<>( @@ -1639,7 +1660,8 @@ public synchronized Pair tableColumnCheckForColumnRenaming( final String newName, final boolean isTableView) throws MetadataException { - final TsTable originalTable = getTableIfExists(database, tableName).orElse(null); + final TsTable originalTable = + clusterSchemaInfo.getTableForModification(database, tableName, oldName, newName); if (Objects.isNull(originalTable)) { return new Pair<>( @@ -1692,7 +1714,7 @@ public synchronized Pair tableCheckForRenaming( final String newName, final boolean isTableView) throws MetadataException { - final TsTable originalTable = getTableIfExists(database, tableName).orElse(null); + final TsTable originalTable = clusterSchemaInfo.getTableForModification(database, tableName); if (Objects.isNull(originalTable)) { return new Pair<>( @@ -1708,7 +1730,12 @@ public synchronized Pair tableCheckForRenaming( return result.get(); } - if (getTableIfExists(database, newName).isPresent()) { + final Optional> targetTable = + getTableAndStatusIfExists(database, newName); + if (targetTable.isPresent() && targetTable.get().getRight() == TableNodeStatus.PRE_DELETE) { + throw new TableInDeletionException(database, newName); + } + if (targetTable.isPresent()) { return new Pair<>( RpcUtils.getStatus( TSStatusCode.TABLE_ALREADY_EXISTS, @@ -1777,7 +1804,7 @@ public synchronized Pair updateTableProperties( final Map updatedProperties, final boolean isTableView) throws MetadataException { - final TsTable originalTable = getTableIfExists(database, tableName).orElse(null); + final TsTable originalTable = clusterSchemaInfo.getTableForModification(database, tableName); if (Objects.isNull(originalTable)) { return new Pair<>( diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java index 772f46baa316..f2d6ae809267 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java @@ -134,6 +134,7 @@ import org.apache.iotdb.confignode.consensus.request.write.table.RenameTableColumnPlan; import org.apache.iotdb.confignode.consensus.request.write.table.RenameTablePlan; import org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan; +import org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreAlterColumnDataTypePlan; import org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreDeleteTablePlan; import org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan; import org.apache.iotdb.confignode.consensus.request.write.table.SetTableCommentPlan; @@ -615,6 +616,9 @@ public TSStatus executeNonQueryPlan(ConfigPhysicalPlan physicalPlan) return clusterSchemaInfo.dropTable((CommitDeleteTablePlan) physicalPlan); case PreAlterColumnDataType: return clusterSchemaInfo.preAlterColumnDataType((PreAlterColumnDataTypePlan) physicalPlan); + case RollbackPreAlterColumnDataType: + return clusterSchemaInfo.rollbackPreAlterColumnDataType( + (RollbackPreAlterColumnDataTypePlan) physicalPlan); case AlterColumnDataType: return clusterSchemaInfo.commitAlterColumnDataType( ((AlterColumnDataTypePlan) physicalPlan)); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java index 8ca27271313f..dc8106a76a60 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java @@ -27,6 +27,7 @@ import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.exception.table.ColumnInAlterException; import org.apache.iotdb.commons.exception.table.ColumnInDeletionException; import org.apache.iotdb.commons.exception.table.TableInDeletionException; import org.apache.iotdb.commons.path.PartialPath; @@ -70,6 +71,7 @@ import org.apache.iotdb.confignode.consensus.request.write.table.RenameTableColumnPlan; import org.apache.iotdb.confignode.consensus.request.write.table.RenameTablePlan; import org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan; +import org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreAlterColumnDataTypePlan; import org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreDeleteTablePlan; import org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan; import org.apache.iotdb.confignode.consensus.request.write.table.SetTableCommentPlan; @@ -112,6 +114,7 @@ import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.annotations.TableModel; +import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -1562,10 +1565,8 @@ public Optional> getTsTableIfExists( } } - public TsTable getTableForColumnExtension( - final String database, - final String tableName, - final List columnSchemaList) + public TsTable getTableForModification( + final String database, final String tableName, final String... columnNames) throws MetadataException { databaseReadWriteLock.readLock().lock(); try { @@ -1580,9 +1581,12 @@ public TsTable getTableForColumnExtension( } final TableSchemaDetails details = tableModelMTree.getTableSchemaDetails(databasePath, tableName); - for (final TsTableColumnSchema column : columnSchemaList) { - if (details.preDeletedColumns.contains(column.getColumnName())) { - throw new ColumnInDeletionException(database, tableName, column.getColumnName()); + for (final String columnName : columnNames) { + if (details.preDeletedColumns.contains(columnName)) { + throw new ColumnInDeletionException(database, tableName, columnName); + } + if (details.preAlteredColumns.containsKey(columnName)) { + throw new ColumnInAlterException(database, tableName, columnName); } } return tableModelMTree.getTableSchemaForDataNode(databasePath, tableName); @@ -1591,6 +1595,50 @@ public TsTable getTableForColumnExtension( } } + public boolean isColumnAlterCommitted( + final String database, + final String tableName, + final String columnName, + final TSDataType dataType) + throws MetadataException { + databaseReadWriteLock.readLock().lock(); + try { + final PartialPath databasePath = getQualifiedDatabasePartialPath(database); + final Optional> tableAndStatus = + tableModelMTree.getTableAndStatusIfExists(databasePath, tableName); + if (!tableAndStatus.isPresent()) { + return false; + } + final TableSchemaDetails details = + tableModelMTree.getTableSchemaDetails(databasePath, tableName); + final TsTableColumnSchema columnSchema = details.table.getColumnSchema(columnName); + return !details.preAlteredColumns.containsKey(columnName) + && columnSchema != null + && columnSchema.getDataType() == dataType; + } finally { + databaseReadWriteLock.readLock().unlock(); + } + } + + public Optional getPreAlteredColumnType( + final String database, final String tableName, final String columnName) + throws MetadataException { + databaseReadWriteLock.readLock().lock(); + try { + final PartialPath databasePath = getQualifiedDatabasePartialPath(database); + if (!tableModelMTree.getTableAndStatusIfExists(databasePath, tableName).isPresent()) { + return Optional.empty(); + } + return Optional.ofNullable( + tableModelMTree + .getTableSchemaDetails(databasePath, tableName) + .preAlteredColumns + .get(columnName)); + } finally { + databaseReadWriteLock.readLock().unlock(); + } + } + public TSStatus addTableColumn(final AddTableColumnPlan plan) { return executeWithLock( () -> { @@ -1678,6 +1726,23 @@ public TSStatus preAlterColumnDataType(final PreAlterColumnDataTypePlan plan) { } } + public TSStatus rollbackPreAlterColumnDataType(final RollbackPreAlterColumnDataTypePlan plan) { + databaseReadWriteLock.writeLock().lock(); + try { + tableModelMTree.rollbackPreAlterColumnDataType( + getQualifiedDatabasePartialPath(plan.getDatabase()), + plan.getTableName(), + plan.getColumnName(), + plan.getNewType()); + return RpcUtils.SUCCESS_STATUS; + } catch (final MetadataException e) { + LOGGER.warn(e.getMessage(), e); + return RpcUtils.getStatus(e.getErrorCode(), e.getMessage()); + } finally { + databaseReadWriteLock.writeLock().unlock(); + } + } + public TSStatus commitAlterColumnDataType(AlterColumnDataTypePlan plan) { databaseReadWriteLock.writeLock().lock(); try { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java index 8b6b8ca8452c..53dacbd7196e 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java @@ -25,6 +25,8 @@ import org.apache.iotdb.commons.exception.IoTDBException; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.exception.table.ColumnInAlterException; +import org.apache.iotdb.commons.exception.table.ColumnInDeletionException; import org.apache.iotdb.commons.exception.table.ColumnNotExistsException; import org.apache.iotdb.commons.exception.table.TableAlreadyExistsException; import org.apache.iotdb.commons.exception.table.TableInDeletionException; @@ -701,6 +703,10 @@ public void preCreateTableView( final IConfigMNode databaseNode = getDatabaseNodeByDatabasePath(database).getAsMNode(); final IConfigMNode node = databaseNode.getChild(table.getTableName()); if (Objects.nonNull(node)) { + if (node instanceof ConfigTableNode + && ((ConfigTableNode) node).getStatus() == TableNodeStatus.PRE_DELETE) { + throw new TableInDeletionException(database.getFullPath(), table.getTableName()); + } if (!TreeViewSchema.isTreeViewTable(((ConfigTableNode) node).getTable())) { throw new TableAlreadyExistsException( database.getFullPath().substring(ROOT.length() + 1), table.getTableName()); @@ -778,7 +784,7 @@ public void dropTable(final PartialPath database, final String tableName) public void renameTable(final PartialPath database, final String tableName, final String newName) throws MetadataException { final IConfigMNode databaseNode = getDatabaseNodeByDatabasePath(database).getAsMNode(); - final ConfigTableNode tableNode = (ConfigTableNode) databaseNode.getChild(tableName); + final ConfigTableNode tableNode = getTableNodeForModification(database, tableName); store.deleteChild(databaseNode, tableName); tableNode.setName(newName); store.addChild(databaseNode, newName, tableNode); @@ -790,7 +796,19 @@ public void renameTableColumn( final String oldName, final String newName) throws MetadataException { - final ConfigTableNode tableNode = getTableNode(database, tableName); + final ConfigTableNode tableNode = getTableNodeForModification(database, tableName); + if (tableNode.getPreDeletedColumns().contains(oldName)) { + throw new ColumnInDeletionException(database.getFullPath(), tableName, oldName); + } + if (tableNode.getPreDeletedColumns().contains(newName)) { + throw new ColumnInDeletionException(database.getFullPath(), tableName, newName); + } + if (tableNode.getPreAlteredColumns().containsKey(oldName)) { + throw new ColumnInAlterException(database.getFullPath(), tableName, oldName); + } + if (tableNode.getPreAlteredColumns().containsKey(newName)) { + throw new ColumnInAlterException(database.getFullPath(), tableName, newName); + } tableNode.getTable().renameColumnSchema(oldName, newName); } @@ -800,7 +818,7 @@ public void setTableComment( final String comment, final boolean isView) throws MetadataException { - final TsTable table = getTable(database, tableName); + final TsTable table = getTableForModification(database, tableName); final Optional> check = ClusterSchemaManager.checkTable4View(database.getTailNode(), table, isView); if (check.isPresent()) { @@ -820,7 +838,8 @@ public void setTableColumnComment( final @Nonnull String columnName, final @Nullable String comment) throws MetadataException { - final TsTable table = getTable(database, tableName); + final ConfigTableNode node = getTableNodeForModification(database, tableName); + final TsTable table = node.getTable(); final TsTableColumnSchema columnSchema = table.getColumnSchema(columnName); @@ -828,6 +847,12 @@ public void setTableColumnComment( throw new ColumnNotExistsException( PathUtils.unQualifyDatabaseName(database.getFullPath()), tableName, columnName); } + if (node.getPreDeletedColumns().contains(columnName)) { + throw new ColumnInDeletionException(database.getFullPath(), tableName, columnName); + } + if (node.getPreAlteredColumns().containsKey(columnName)) { + throw new ColumnInAlterException(database.getFullPath(), tableName, columnName); + } if (Objects.nonNull(comment)) { columnSchema.getProps().put(TsTable.COMMENT_KEY, comment); } else { @@ -943,7 +968,13 @@ public void addTableColumn( final String tableName, final List columnSchemaList) throws MetadataException { - final TsTable table = getTable(database, tableName); + final TsTable table = + getTableForModification( + database, + tableName, + columnSchemaList.stream() + .map(TsTableColumnSchema::getColumnName) + .toArray(String[]::new)); columnSchemaList.forEach(table::addColumnSchema); } @@ -964,7 +995,8 @@ public void setTableProperties( throw new TableNotExistsException( database.getFullPath().substring(ROOT.length() + 1), tableName); } - final TsTable table = ((ConfigTableNode) databaseNode.getChild(tableName)).getTable(); + final ConfigTableNode tableNode = getTableNodeForModification(database, tableName); + final TsTable table = tableNode.getTable(); tableProperties.forEach( (k, v) -> { if (Objects.nonNull(v)) { @@ -992,7 +1024,7 @@ public boolean preDeleteColumn( final String columnName, final boolean isView) throws MetadataException, SemanticException { - final ConfigTableNode node = getTableNode(database, tableName); + final ConfigTableNode node = getTableNodeForModification(database, tableName); final Optional> check = ClusterSchemaManager.checkTable4View(database.getTailNode(), node.getTable(), isView); if (check.isPresent()) { @@ -1010,6 +1042,10 @@ public boolean preDeleteColumn( throw new SemanticException(ConfigNodeMessages.DROPPING_TAG_OR_TIME_COLUMN_IS_NOT_SUPPORTED); } + if (node.getPreAlteredColumns().containsKey(columnName)) { + throw new ColumnInAlterException(database.getFullPath(), tableName, columnName); + } + node.addPreDeletedColumn(columnName); return columnSchema.getColumnCategory() == TsTableColumnCategory.ATTRIBUTE; } @@ -1022,22 +1058,34 @@ public void commitDeleteColumn( if (Objects.nonNull(table.getColumnSchema(columnName))) { table.removeColumnSchema(columnName); node.removePreDeletedColumn(columnName); + node.removePreAlteredColumn(columnName); } } public void preAlterColumnDataType( PartialPath database, String tableName, String columnName, TSDataType dataType) throws MetadataException { - final ConfigTableNode node = getTableNode(database, tableName); + final ConfigTableNode node = getTableNodeForModification(database, tableName); final TsTableColumnSchema columnSchema = node.getTable().getColumnSchema(columnName); if (Objects.isNull(columnSchema)) { throw new ColumnNotExistsException( PathUtils.unQualifyDatabaseName(database.getFullPath()), tableName, columnName); } + if (node.getPreDeletedColumns().contains(columnName)) { + throw new ColumnInDeletionException(database.getFullPath(), tableName, columnName); + } if (columnSchema.getColumnCategory() != TsTableColumnCategory.FIELD) { throw new SemanticException(ConfigNodeMessages.CAN_ONLY_ALTER_DATATYPE_OF_FIELD_COLUMNS); } + if (node.getPreAlteredColumns().containsKey(columnName)) { + final TSDataType currentType = node.getPreAlteredColumns().get(columnName); + if (currentType == dataType) { + return; + } + throw new ColumnInAlterException(database.getFullPath(), tableName, columnName); + } + if (!MetadataUtils.canAlter(columnSchema.getDataType(), dataType)) { throw new SemanticException( String.format( @@ -1052,8 +1100,19 @@ public void preAlterColumnDataType( public void commitAlterColumnDataType( PartialPath database, String tableName, String columnName, TSDataType dataType) throws MetadataException { - final ConfigTableNode node = getTableNode(database, tableName); - final TsTable table = getTable(database, tableName); + final IConfigMNode databaseNode = getDatabaseNodeByDatabasePath(database).getAsMNode(); + if (!databaseNode.hasChild(tableName)) { + return; + } + final IConfigMNode tableNode = databaseNode.getChild(tableName); + if (!(tableNode instanceof ConfigTableNode)) { + return; + } + final ConfigTableNode node = (ConfigTableNode) tableNode; + if (!Objects.equals(node.getPreAlteredColumns().get(columnName), dataType)) { + return; + } + final TsTable table = node.getTable(); final TsTableColumnSchema columnSchema = table.getColumnSchema(columnName); if (Objects.nonNull(columnSchema)) { columnSchema.setDataType(dataType); @@ -1062,6 +1121,26 @@ public void commitAlterColumnDataType( fieldColumnSchema.setEncoding( SchemaUtils.getDataTypeCompatibleEncoding(dataType, fieldColumnSchema.getEncoding())); } + } + node.removePreAlteredColumn(columnName); + } + + public void rollbackPreAlterColumnDataType( + final PartialPath database, + final String tableName, + final String columnName, + final TSDataType dataType) + throws MetadataException { + final IConfigMNode databaseNode = getDatabaseNodeByDatabasePath(database).getAsMNode(); + if (!databaseNode.hasChild(tableName)) { + return; + } + final IConfigMNode tableNode = databaseNode.getChild(tableName); + if (!(tableNode instanceof ConfigTableNode)) { + return; + } + final ConfigTableNode node = (ConfigTableNode) tableNode; + if (Objects.equals(node.getPreAlteredColumns().get(columnName), dataType)) { node.removePreAlteredColumn(columnName); } } @@ -1072,13 +1151,28 @@ public TsTable getTableSchemaForDataNode(final PartialPath database, final Strin } private TsTable getTableSchemaForDataNode(final ConfigTableNode node) { - if (node.getPreDeletedColumns().isEmpty()) { + if (node.getPreDeletedColumns().isEmpty() && node.getPreAlteredColumns().isEmpty()) { return node.getTable(); } // Cache reloads and later schema updates must not make a column writable again while its // deletion is still pending. DESC uses the complete schema separately. final TsTable table = new TsTable(node.getTable()); node.getPreDeletedColumns().forEach(table::removeColumnSchema); + node.getPreAlteredColumns() + .forEach( + (columnName, dataType) -> { + final TsTableColumnSchema columnSchema = table.getColumnSchema(columnName); + if (columnSchema == null) { + return; + } + columnSchema.setDataType(dataType); + if (columnSchema instanceof FieldColumnSchema) { + final FieldColumnSchema fieldColumnSchema = (FieldColumnSchema) columnSchema; + fieldColumnSchema.setEncoding( + SchemaUtils.getDataTypeCompatibleEncoding( + dataType, fieldColumnSchema.getEncoding())); + } + }); return table; } @@ -1127,6 +1221,30 @@ private TsTable getTable(final PartialPath database, final String tableName) return getTableNode(database, tableName).getTable(); } + private TsTable getTableForModification( + final PartialPath database, final String tableName, final String... columnNames) + throws MetadataException { + final ConfigTableNode node = getTableNodeForModification(database, tableName); + for (final String columnName : columnNames) { + if (node.getPreDeletedColumns().contains(columnName)) { + throw new ColumnInDeletionException(database.getFullPath(), tableName, columnName); + } + if (node.getPreAlteredColumns().containsKey(columnName)) { + throw new ColumnInAlterException(database.getFullPath(), tableName, columnName); + } + } + return node.getTable(); + } + + private ConfigTableNode getTableNodeForModification( + final PartialPath database, final String tableName) throws MetadataException { + final ConfigTableNode node = getTableNode(database, tableName); + if (node.getStatus() == TableNodeStatus.PRE_DELETE) { + throw new TableInDeletionException(database.getFullPath(), tableName); + } + return node; + } + public Optional> getTableAndStatusIfExists( final PartialPath database, final String tableName) throws MetadataException { final IConfigMNode databaseNode = getDatabaseNodeByDatabasePath(database).getAsMNode(); 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 12491f24a486..dd63a1ecaf0a 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 @@ -24,9 +24,11 @@ import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.confignode.consensus.request.write.table.AlterColumnDataTypePlan; +import org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreAlterColumnDataTypePlan; import org.apache.iotdb.confignode.i18n.ProcedureMessages; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.confignode.procedure.exception.ProcedureException; +import org.apache.iotdb.confignode.procedure.impl.schema.SchemaUtils; import org.apache.iotdb.confignode.procedure.state.schema.AlterTableColumnDataTypeState; import org.apache.iotdb.confignode.procedure.store.ProcedureType; import org.apache.iotdb.rpc.TSStatusCode; @@ -41,6 +43,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.Objects; +import java.util.Optional; public class AlterTableColumnDataTypeProcedure extends AbstractAlterOrDropTableProcedure { @@ -156,14 +159,25 @@ private void alterColumnDataType(final ConfigNodeProcedureEnv env) { new AlterColumnDataTypePlan(database, tableName, columnName, dataType), isGeneratedByPipe); if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + // A consensus write may have been applied even when the client observed a failure (for + // example, a timeout after the leader committed the entry). Continue with release when the + // canonical CN schema already contains the requested type; rolling back in that case would + // leave DataNodes on an older schema than the CN. + if (isColumnAlterCommitted(env)) { + setNextState(AlterTableColumnDataTypeState.COMMIT_RELEASE); + return; + } setFailure(new ProcedureException(new IoTDBException(status.getMessage(), status.getCode()))); + return; } setNextState(AlterTableColumnDataTypeState.COMMIT_RELEASE); } @Override protected boolean isRollbackSupported(final AlterTableColumnDataTypeState state) { - return false; + return state == AlterTableColumnDataTypeState.CHECK_AND_INVALIDATE_COLUMN + || state == AlterTableColumnDataTypeState.PRE_RELEASE + || state == AlterTableColumnDataTypeState.ALTER_TABLE_COLUMN_DATA_TYPE; } @Override @@ -171,7 +185,67 @@ protected void rollbackState( final ConfigNodeProcedureEnv configNodeProcedureEnv, final AlterTableColumnDataTypeState alterTableColumnDataTypeState) throws IOException, InterruptedException, ProcedureException { - // Do nothing + // COMMIT_RELEASE is irreversible: the CN schema has already been committed and must not be + // followed by a cache rollback if the procedure is aborted while notifying DataNodes. + if (alterTableColumnDataTypeState == AlterTableColumnDataTypeState.COMMIT_RELEASE) { + return; + } + final Optional pendingType = getPreAlteredColumnType(configNodeProcedureEnv); + if (pendingType.isPresent() && pendingType.get() != dataType) { + // This rollback belongs to an older procedure. Leave a newer pre-alter marker and its + // DataNode cache entry untouched. + return; + } + final boolean ownsPendingMarker = pendingType.isPresent(); + if (!ownsPendingMarker && isColumnAlterCommittedForRollback(configNodeProcedureEnv)) { + return; + } + final TSStatus status = + SchemaUtils.executeInConsensusLayer( + new RollbackPreAlterColumnDataTypePlan(database, tableName, columnName, dataType), + configNodeProcedureEnv, + LOGGER); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new ProcedureException(new IoTDBException(status.getMessage(), status.getCode())); + } + if (alterTableColumnDataTypeState != AlterTableColumnDataTypeState.CHECK_AND_INVALIDATE_COLUMN + && table != null + && !getPreAlteredColumnType(configNodeProcedureEnv).isPresent() + && (ownsPendingMarker || !isColumnAlterCommittedForRollback(configNodeProcedureEnv))) { + rollbackPreRelease(configNodeProcedureEnv); + } + } + + private Optional getPreAlteredColumnType(final ConfigNodeProcedureEnv env) + throws ProcedureException { + try { + return env.getConfigManager() + .getClusterSchemaManager() + .getPreAlteredColumnType(database, tableName, columnName); + } catch (final MetadataException e) { + throw new ProcedureException(e); + } + } + + private boolean isColumnAlterCommittedForRollback(final ConfigNodeProcedureEnv env) + throws ProcedureException { + try { + return env.getConfigManager() + .getClusterSchemaManager() + .isColumnAlterCommitted(database, tableName, columnName, dataType); + } catch (final MetadataException e) { + throw new ProcedureException(e); + } + } + + private boolean isColumnAlterCommitted(final ConfigNodeProcedureEnv env) { + try { + return env.getConfigManager() + .getClusterSchemaManager() + .isColumnAlterCommitted(database, tableName, columnName, dataType); + } catch (final MetadataException e) { + return false; + } } @Override diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java index 0e8585898031..1b9f4d550293 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java @@ -147,6 +147,7 @@ import org.apache.iotdb.confignode.consensus.request.write.table.RenameTableColumnPlan; import org.apache.iotdb.confignode.consensus.request.write.table.RenameTablePlan; import org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan; +import org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreAlterColumnDataTypePlan; import org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan; import org.apache.iotdb.confignode.consensus.request.write.table.SetTableCommentPlan; import org.apache.iotdb.confignode.consensus.request.write.table.SetTablePropertiesPlan; @@ -1716,6 +1717,20 @@ public void PreAlterTableColumnDataTypePlanTest() throws IOException { alterColumnDataTypePlan.getNewType(), alterColumnDataTypePlan1.getNewType()); } + @Test + public void RollbackPreAlterTableColumnDataTypePlanTest() throws IOException { + final RollbackPreAlterColumnDataTypePlan rollbackPlan = + new RollbackPreAlterColumnDataTypePlan("database1", "table1", "field", TSDataType.FLOAT); + final RollbackPreAlterColumnDataTypePlan rollbackPlan1 = + (RollbackPreAlterColumnDataTypePlan) + ConfigPhysicalPlan.Factory.create(rollbackPlan.serializeToByteBuffer()); + Assert.assertEquals(rollbackPlan.getDatabase(), rollbackPlan1.getDatabase()); + Assert.assertEquals(rollbackPlan.getTableName(), rollbackPlan1.getTableName()); + Assert.assertEquals(rollbackPlan.getColumnName(), rollbackPlan1.getColumnName()); + Assert.assertEquals(rollbackPlan.getType(), rollbackPlan1.getType()); + Assert.assertEquals(rollbackPlan.getNewType(), rollbackPlan1.getNewType()); + } + @Test public void AlterTableColumnDataTypePlanTest() throws IOException { final AlterColumnDataTypePlan alterColumnDataTypePlan = diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java index 2ad27efd1153..318c957b825e 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java @@ -49,6 +49,7 @@ import java.io.InputStream; import java.nio.file.Files; import java.util.ArrayList; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -360,6 +361,10 @@ public void testTableSerialization() throws Exception { root.preCreateTable(pathList[i], table); root.commitCreateTable(pathList[i], tableName); + if (i == 0) { + Assert.assertTrue(root.preDeleteColumn(pathList[i], tableName, "Attr", false)); + root.preAlterColumnDataType(pathList[i], tableName, "Measurement", TSDataType.STRING); + } } final ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); @@ -402,8 +407,21 @@ public void testTableSerialization() throws Exception { final TsTable table = tables.get(0); assertEquals("table" + i, table.getTableName()); assertEquals(1, table.getTagNum()); - // currently, only construct the TsTable would not carry the time column - assertEquals(3, table.getColumnNum()); + // These TsTables have no time column; the first table also hides its pre-deleted attribute. + assertEquals(i == 0 ? 2 : 3, table.getColumnNum()); + final ConfigMTree.TableSchemaDetails details = + newTree.getTableSchemaDetails(pathList[i], table.getTableName()); + if (i == 0) { + assertEquals(Collections.singleton("Attr"), details.preDeletedColumns); + assertEquals(TSDataType.STRING, details.preAlteredColumns.get("Measurement")); + assertEquals(TSDataType.DOUBLE, details.table.getColumnSchema("Measurement").getDataType()); + Assert.assertNotNull(details.table.getColumnSchema("Attr")); + assertEquals(TSDataType.STRING, table.getColumnSchema("Measurement").getDataType()); + Assert.assertNull(table.getColumnSchema("Attr")); + } else { + assertTrue(details.preDeletedColumns.isEmpty()); + assertTrue(details.preAlteredColumns.isEmpty()); + } } } @@ -453,6 +471,57 @@ public void testAlterColumnTypeUpdatesCompatibleEncoding() throws Exception { root.getTableSchemaDetails(database, table.getTableName()).preAlteredColumns.isEmpty()); } + @Test + public void testRollbackPreAlterColumnDataTypeOnlyClearsMatchingRequest() throws Exception { + root = new ConfigMTree(true); + + final PartialPath database = new PartialPath("root.sg"); + root.setStorageGroup(database); + final IDatabaseMNode databaseNode = root.getDatabaseNodeByDatabasePath(database); + databaseNode + .getAsMNode() + .getDatabaseSchema() + .setName(PathUtils.unQualifyDatabaseName(database.getFullPath())); + databaseNode.getAsMNode().getDatabaseSchema().setIsTableModel(true); + + final TsTable table = new TsTable("table1"); + table.addColumnSchema(new TagColumnSchema("id", TSDataType.STRING)); + table.addColumnSchema( + new FieldColumnSchema( + "measurement", TSDataType.DOUBLE, TSEncoding.GORILLA, CompressionType.SNAPPY)); + root.preCreateTable(database, table); + root.commitCreateTable(database, table.getTableName()); + + root.preAlterColumnDataType(database, table.getTableName(), "measurement", TSDataType.STRING); + Assert.assertEquals( + TSDataType.STRING, + root.getTableSchemaForDataNode(database, table.getTableName()) + .getColumnSchema("measurement") + .getDataType()); + + // A stale rollback must not clear a newer request for a different target type. + root.rollbackPreAlterColumnDataType( + database, table.getTableName(), "measurement", TSDataType.FLOAT); + Assert.assertEquals( + TSDataType.STRING, + root.getTableSchemaDetails(database, table.getTableName()) + .preAlteredColumns + .get("measurement")); + + root.rollbackPreAlterColumnDataType( + database, table.getTableName(), "measurement", TSDataType.STRING); + Assert.assertTrue( + root.getTableSchemaDetails(database, table.getTableName()).preAlteredColumns.isEmpty()); + // A delayed commit from the failed procedure must not apply after the marker was rolled back. + root.commitAlterColumnDataType( + database, table.getTableName(), "measurement", TSDataType.STRING); + Assert.assertEquals( + TSDataType.DOUBLE, + root.getTableSchemaForDataNode(database, table.getTableName()) + .getColumnSchema("measurement") + .getDataType()); + } + @Test public void testSetTemplate() throws MetadataException { root.setStorageGroup(new PartialPath("root.a")); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java index 10984893d60a..86b74ec00b6b 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java @@ -20,6 +20,7 @@ package org.apache.iotdb.confignode.persistence.schema; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.exception.table.ColumnInAlterException; import org.apache.iotdb.commons.exception.table.ColumnInDeletionException; import org.apache.iotdb.commons.exception.table.TableInDeletionException; import org.apache.iotdb.commons.schema.table.TableNodeStatus; @@ -33,6 +34,8 @@ import org.apache.iotdb.confignode.consensus.request.read.table.FetchTablePlan; import org.apache.iotdb.confignode.consensus.request.read.table.ShowTablePlan; import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan; +import org.apache.iotdb.confignode.consensus.request.write.table.AddTableColumnPlan; +import org.apache.iotdb.confignode.consensus.request.write.table.AlterColumnDataTypePlan; import org.apache.iotdb.confignode.consensus.request.write.table.CommitCreateTablePlan; import org.apache.iotdb.confignode.consensus.request.write.table.CommitDeleteColumnPlan; import org.apache.iotdb.confignode.consensus.request.write.table.CommitDeleteTablePlan; @@ -40,6 +43,7 @@ import org.apache.iotdb.confignode.consensus.request.write.table.PreCreateTablePlan; import org.apache.iotdb.confignode.consensus.request.write.table.PreDeleteColumnPlan; import org.apache.iotdb.confignode.consensus.request.write.table.PreDeleteTablePlan; +import org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan; import org.apache.iotdb.confignode.manager.IManager; import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager; import org.apache.iotdb.confignode.manager.schema.ClusterSchemaQuotaStatistics; @@ -152,6 +156,10 @@ public void testDescRetainsPreDeletedColumnsAndAlteredTypes() { assertEquals(TSDataType.INT64, table.getColumnSchema("live").getDataType()); assertFalse(basic.isSetPreDeletedColumns()); + assertEquals( + TSDataType.INT64, + schemaInfo.getAllUsingTables().get(DATABASE).get(0).getColumnSchema("live").getDataType()); + final TDescTableResp details = describe(true); assertTrue(details.getPreDeletedColumns().containsAll(Arrays.asList("field", "attribute"))); assertEquals( @@ -209,6 +217,67 @@ public void testColumnExtensionRejectsPreDeletedNames() throws Exception { .getLeft()); } + @Test + public void testPreAlterRejectsConflictingColumnOperations() throws Exception { + assertSuccess( + schemaInfo.preAlterColumnDataType( + new PreAlterColumnDataTypePlan(DATABASE, TABLE, "live", TSDataType.INT64))); + // Retrying the same target type is allowed so a stuck procedure can be resumed. + assertSuccess( + schemaInfo.preAlterColumnDataType( + new PreAlterColumnDataTypePlan(DATABASE, TABLE, "live", TSDataType.INT64))); + + final TSStatus secondAlter = + schemaInfo.preAlterColumnDataType( + new PreAlterColumnDataTypePlan(DATABASE, TABLE, "live", TSDataType.FLOAT)); + assertEquals(TSStatusCode.SEMANTIC_ERROR.getStatusCode(), secondAlter.getCode()); + assertEquals( + new ColumnInAlterException(DATABASE, TABLE, "live").getMessage(), secondAlter.getMessage()); + + final TSStatus delete = + schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE, TABLE, "live")); + assertEquals(TSStatusCode.SEMANTIC_ERROR.getStatusCode(), delete.getCode()); + assertEquals( + new ColumnInAlterException(DATABASE, TABLE, "live").getMessage(), delete.getMessage()); + + assertThrows( + ColumnInAlterException.class, + () -> + schemaManager.tableColumnCheckForColumnExtension( + DATABASE, TABLE, new ArrayList<>(Collections.singletonList(field("live"))), false)); + assertEquals( + new ColumnInAlterException(DATABASE, TABLE, "live").getMessage(), + schemaInfo + .addTableColumn( + new AddTableColumnPlan( + DATABASE, TABLE, Collections.singletonList(field("live")), false)) + .getMessage()); + assertEquals( + new ColumnInAlterException(DATABASE, TABLE, "live").getMessage(), + schemaInfo + .setTableColumnComment( + new SetTableColumnCommentPlan(DATABASE, TABLE, "live", "comment")) + .getMessage()); + assertThrows( + ColumnInAlterException.class, + () -> + schemaManager.tableColumnCheckForColumnRenaming( + DATABASE, TABLE, "live", "renamed", false)); + } + + @Test + public void testSameTypePreAlterIsNotReportedAsCommittedUntilMarkerIsCleared() throws Exception { + assertSuccess( + schemaInfo.preAlterColumnDataType( + new PreAlterColumnDataTypePlan(DATABASE, TABLE, "live", TSDataType.INT32))); + assertFalse(schemaInfo.isColumnAlterCommitted(DATABASE, TABLE, "live", TSDataType.INT32)); + + assertSuccess( + schemaInfo.commitAlterColumnDataType( + new AlterColumnDataTypePlan(DATABASE, TABLE, "live", TSDataType.INT32))); + assertTrue(schemaInfo.isColumnAlterCommitted(DATABASE, TABLE, "live", TSDataType.INT32)); + } + @Test public void testPreDeletedTableRejectsCreationAndColumnExtension() { assertSuccess(schemaInfo.preDeleteTable(new PreDeleteTablePlan(DATABASE, TABLE))); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedureTest.java new file mode 100644 index 000000000000..5634abbe6bf9 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedureTest.java @@ -0,0 +1,171 @@ +/* + * 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.schema.table; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.consensus.ConsensusManager; +import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.state.schema.AlterTableColumnDataTypeState; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.tsfile.enums.TSDataType; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import java.lang.reflect.Method; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; + +public class AlterTableColumnDataTypeProcedureTest { + + @Test + public void commitReleaseIsNeverRolledBack() throws Exception { + final AlterTableColumnDataTypeProcedure procedure = + new AlterTableColumnDataTypeProcedure("database", "table", "query", "value", null, false); + procedure.table = new TsTable("table"); + + // The commit state may be present on the rollback stack when an abort races with cache + // notification. It must not issue a rollback plan or touch DataNode caches. + procedure.rollbackState(null, AlterTableColumnDataTypeState.COMMIT_RELEASE); + } + + @Test + public void rollbackClearsCnMarkerAfterProcedureRestartBeforeTableSnapshot() throws Exception { + final ConsensusManager consensusManager = Mockito.mock(ConsensusManager.class); + final ClusterSchemaManager schemaManager = Mockito.mock(ClusterSchemaManager.class); + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + Mockito.when(env.getConfigManager()).thenReturn(configManager); + Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager); + Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager); + Mockito.when(consensusManager.write(Mockito.any())) + .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + Mockito.when( + schemaManager.isColumnAlterCommitted("database", "table", "value", TSDataType.INT64)) + .thenReturn(false); + + final AlterTableColumnDataTypeProcedure procedure = + new AlterTableColumnDataTypeProcedure( + "database", "table", "query", "value", TSDataType.INT64, false); + + // The table snapshot is intentionally null, as it can be after a restart that happened after + // the pre-alter consensus entry was applied but before the procedure persisted its snapshot. + procedure.rollbackState(env, AlterTableColumnDataTypeState.CHECK_AND_INVALIDATE_COLUMN); + + Mockito.verify(consensusManager).write(Mockito.any()); + } + + @Test + public void sameTypePreAlterRollbackStillCleansDataNodeCache() throws Exception { + final ConsensusManager consensusManager = Mockito.mock(ConsensusManager.class); + final ClusterSchemaManager schemaManager = Mockito.mock(ClusterSchemaManager.class); + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + Mockito.when(env.getConfigManager()).thenReturn(configManager); + Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager); + Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager); + + final AtomicBoolean markerCleared = new AtomicBoolean(false); + Mockito.when(schemaManager.getPreAlteredColumnType("database", "table", "value")) + .thenAnswer( + invocation -> markerCleared.get() ? Optional.empty() : Optional.of(TSDataType.INT64)); + Mockito.when( + schemaManager.isColumnAlterCommitted("database", "table", "value", TSDataType.INT64)) + .thenAnswer(invocation -> markerCleared.get()); + Mockito.when(consensusManager.write(Mockito.any())) + .thenAnswer( + invocation -> { + markerCleared.set(true); + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + }); + + final TestAlterTableColumnDataTypeProcedure procedure = + new TestAlterTableColumnDataTypeProcedure(); + procedure.table = new TsTable("table"); + procedure.rollbackState(env, AlterTableColumnDataTypeState.PRE_RELEASE); + + Assert.assertTrue(procedure.dataNodeRollbackCalled); + } + + @Test + public void staleRollbackDoesNotClearNewerPreAlter() throws Exception { + final ConsensusManager consensusManager = Mockito.mock(ConsensusManager.class); + final ClusterSchemaManager schemaManager = Mockito.mock(ClusterSchemaManager.class); + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + Mockito.when(env.getConfigManager()).thenReturn(configManager); + Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager); + Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager); + Mockito.when(schemaManager.getPreAlteredColumnType("database", "table", "value")) + .thenReturn(Optional.of(TSDataType.FLOAT)); + + final TestAlterTableColumnDataTypeProcedure procedure = + new TestAlterTableColumnDataTypeProcedure(); + procedure.table = new TsTable("table"); + procedure.rollbackState(env, AlterTableColumnDataTypeState.PRE_RELEASE); + + Mockito.verify(consensusManager, Mockito.never()).write(Mockito.any()); + Assert.assertFalse(procedure.dataNodeRollbackCalled); + } + + @Test + public void consensusFailureAfterCommitContinuesToCommitRelease() throws Exception { + final ClusterSchemaManager schemaManager = Mockito.mock(ClusterSchemaManager.class); + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + Mockito.when(env.getConfigManager()).thenReturn(configManager); + Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager); + + Mockito.when(schemaManager.executePlan(Mockito.any(), Mockito.eq(false))) + .thenReturn(new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())); + Mockito.when( + schemaManager.isColumnAlterCommitted("database", "table", "value", TSDataType.INT64)) + .thenReturn(true); + + final AlterTableColumnDataTypeProcedure procedure = + new AlterTableColumnDataTypeProcedure( + "database", "table", "query", "value", TSDataType.INT64, false); + final Method alterColumnDataType = + AlterTableColumnDataTypeProcedure.class.getDeclaredMethod( + "alterColumnDataType", ConfigNodeProcedureEnv.class); + alterColumnDataType.setAccessible(true); + alterColumnDataType.invoke(procedure, env); + + Assert.assertFalse(procedure.isFailed()); + } + + private static class TestAlterTableColumnDataTypeProcedure + extends AlterTableColumnDataTypeProcedure { + private boolean dataNodeRollbackCalled; + + private TestAlterTableColumnDataTypeProcedure() { + super("database", "table", "query", "value", TSDataType.INT64, false); + } + + @Override + protected void rollbackPreRelease(final ConfigNodeProcedureEnv env) { + dataNodeRollbackCalled = true; + } + } +} 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 88b09d190747..39520259da1c 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 @@ -88,6 +88,13 @@ public class DataNodeTableCache implements ITableCache { private final Map>> specialStatusMap = new ConcurrentHashMap<>(); + /** + * The cache entry replaced by the latest pre-update. It lets a rollback restore a schema that may + * already have been promoted from {@code specialStatusMap} by a concurrent fetch. A {@link + * NonCommittableTsTable} value means that the previous schema is unavailable after a restart. + */ + private final Map> previousTableMap = new ConcurrentHashMap<>(); + private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock(); private final Semaphore fetchTableSemaphore = new Semaphore( @@ -120,6 +127,7 @@ public void init(final byte[] tableInitializationBytes) { TsTableInternalRPCUtil.deserializeTableInitializationInfo(tableInitializationBytes); final Map> usingMap = tableInfo.left; final Map> specialStatusMap = tableInfo.right; + previousTableMap.clear(); usingMap.forEach( (key, value) -> databaseTableMap.put( @@ -142,6 +150,19 @@ public void init(final byte[] tableInitializationBytes) { table -> new Pair<>(table, 0L), (v1, v2) -> v2, ConcurrentHashMap::new)))); + specialStatusMap.forEach( + (key, value) -> + value.stream() + .filter(NonCommittableTsTable.class::isInstance) + .forEach( + table -> + previousTableMap + .computeIfAbsent( + PathUtils.unQualifyDatabaseName(key), + database -> new ConcurrentHashMap<>()) + .put( + table.getTableName(), + new NonCommittableTsTable(table.getTableName())))); LOGGER.info(DataNodeSchemaMessages.INIT_TABLE_CACHE_SUCCESS); } finally { readWriteLock.writeLock().unlock(); @@ -184,6 +205,14 @@ public void preUpdateTable(String database, final TsTable table, final String ol readWriteLock.writeLock().lock(); try { failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS); + if (oldName == null && !(table instanceof PreDeleteTsTable)) { + final TsTable previousTable = getTableFromCache(database, table.getTableName()); + if (previousTable != null) { + previousTableMap + .computeIfAbsent(database, k -> new ConcurrentHashMap<>()) + .putIfAbsent(table.getTableName(), new TsTable(previousTable)); + } + } specialStatusMap .computeIfAbsent(database, k -> new ConcurrentHashMap<>()) .compute( @@ -240,13 +269,47 @@ public void rollbackUpdateTable(String database, final String tableName, final S failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS); // if rollback the drop table procedure, do nothing, // wait for triggering the action of pull table from CN - final TsTable table = getTableFromSpecialStatusMap(database, tableName); + final Map> databaseSpecialStatusMap = + specialStatusMap.get(database); + final Pair tableStatusPair = + databaseSpecialStatusMap == null ? null : databaseSpecialStatusMap.get(tableName); + final TsTable table = tableStatusPair == null ? null : tableStatusPair.getLeft(); + final TsTable previousTable = + previousTableMap.containsKey(database) + ? previousTableMap.get(database).get(tableName) + : null; if (table instanceof PreDeleteTsTable) { + removePreviousTable(database, tableName); + return; + } + // A null pending value with no saved previous schema means commit already consumed this + // update. A delayed rollback must not evict the committed table. + if (Objects.isNull(oldName) && table == null && previousTable == null) { return; } removeTableFromSpecialStatusMap(database, tableName); + removePreviousTable(database, tableName); LOGGER.info(DataNodeSchemaMessages.ROLLBACK_UPDATE_TABLE_SUCCESS, database, tableName); + // A table fetched while the update was pending can already be in databaseTableMap and the + // special entry can consequently have a null left value. Restore the snapshot captured at + // PRE_UPDATE time, or evict the entry so the next lookup must fetch the canonical CN schema. + if (Objects.isNull(oldName) && tableStatusPair != null) { + if (previousTable != null && !(previousTable instanceof NonCommittableTsTable)) { + databaseTableMap + .computeIfAbsent(database, k -> new ConcurrentHashMap<>()) + .put(tableName, previousTable); + } else if (databaseTableMap.containsKey(database)) { + databaseTableMap.get(database).remove(tableName); + } + if (previousTable == null || previousTable instanceof NonCommittableTsTable) { + // The previous schema is unavailable after a restart or when this was a newly-created + // table. Keep a non-committable marker so getTable() fetches the canonical CN state + // instead of serving a stale entry or permanently treating the cache as already handled. + tableStatusPair.setLeft(new NonCommittableTsTable(tableName)); + } + } + // If rename table if (Objects.nonNull(oldName)) { // Equals to commit update @@ -314,6 +377,15 @@ private void removeTableFromSpecialStatusMap(final String database, final String }); } + private void removePreviousTable(final String database, final String tableName) { + previousTableMap.computeIfPresent( + database, + (k, v) -> { + v.remove(tableName); + return v.isEmpty() ? null : v; + }); + } + @Override public void commitUpdateTable( String database, final String tableName, final @Nullable String oldName) { @@ -331,6 +403,7 @@ public void commitUpdateTable( if (Objects.nonNull(oldName)) { removeTableFromSpecialStatusMap(database, oldName); } + removePreviousTable(database, tableName); return; } // Cannot be committed, consider: @@ -360,6 +433,7 @@ public void commitUpdateTable( LOGGER.info(DataNodeSchemaMessages.COMMIT_UPDATE_TABLE_SUCCESS, database, tableName); } removeTableFromSpecialStatusMap(database, tableName); + removePreviousTable(database, tableName); if (Objects.nonNull(oldName)) { removeTableFromSpecialStatusMap(database, oldName); LOGGER.info(DataNodeSchemaMessages.RENAME_OLD_TABLE_SUCCESS, database, oldName); @@ -375,6 +449,7 @@ private void commitDeleteTable(String database, final String tableName) { databaseTableMap.get(database).remove(tableName); } removeTableFromSpecialStatusMap(database, tableName); + removePreviousTable(database, tableName); LOGGER.info(DataNodeSchemaMessages.COMMIT_DELETE_TABLE_SUCCESS, database, tableName); } @@ -385,6 +460,7 @@ public void invalid(String database) { try { databaseTableMap.remove(database); specialStatusMap.remove(database); + previousTableMap.remove(database); instanceVersion.incrementAndGet(); } finally { readWriteLock.writeLock().unlock(); @@ -402,6 +478,7 @@ public void invalidateAll() { try { databaseTableMap.clear(); specialStatusMap.clear(); + previousTableMap.clear(); instanceVersion.incrementAndGet(); } finally { readWriteLock.writeLock().unlock(); 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 217532f6c99e..78af8f3eee0e 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 @@ -22,8 +22,10 @@ import org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy; import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.exception.table.TableInDeletionException; +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.tsfile.enums.TSDataType; @@ -36,6 +38,7 @@ import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.Collections; +import java.util.HashMap; import java.util.Map; import java.util.concurrent.Semaphore; @@ -43,6 +46,7 @@ public class DataNodeTableCacheTest { private static final String DATABASE = "interrupted_fetch_database"; private static final String TABLE_CACHE_TEST_DATABASE = "root.table_cache_test"; + private static final String TABLE_CACHE_TEST_DATABASE_NAME = "table_cache_test"; private static final String TABLE_NAME = "table1"; @Test @@ -93,7 +97,99 @@ public void commitAfterRollbackUpdateTableIsIgnored() { cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); - Assert.assertNull(cache.getTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, false)); + Assert.assertFalse( + cache + .getTableSnapshot() + .getOrDefault(TABLE_CACHE_TEST_DATABASE_NAME, Collections.emptyMap()) + .containsKey(TABLE_NAME)); + } finally { + cache.invalid(TABLE_CACHE_TEST_DATABASE); + } + } + + @Test + public void rollbackAlteredTableRestoresOriginalSchema() throws Exception { + final ITableCache cache = DataNodeTableCache.getInstance(); + cache.invalid(TABLE_CACHE_TEST_DATABASE); + try { + cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, createTable(TABLE_NAME), null); + cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + + final TsTable alteredTable = createTable(TABLE_NAME); + ((FieldColumnSchema) alteredTable.getColumnSchema("s1")).setDataType(TSDataType.DOUBLE); + cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, alteredTable, null); + + // A concurrent fetch may promote the pending table into the regular cache before rollback. + // Keep that path covered because rollback must still restore the pre-update schema. + final Method updateUsingTable = + DataNodeTableCache.class.getDeclaredMethod( + "updateUsingTable", Map.class, Map.class, LeaseFencedRetryPolicy.class); + updateUsingTable.setAccessible(true); + final Map> fetchedTables = new HashMap<>(); + fetchedTables.put( + TABLE_CACHE_TEST_DATABASE, Collections.singletonMap(TABLE_NAME, alteredTable)); + final Map> previousVersions = new HashMap<>(); + previousVersions.put( + TABLE_CACHE_TEST_DATABASE_NAME, Collections.singletonMap(TABLE_NAME, 1L)); + updateUsingTable.invoke( + cache, fetchedTables, previousVersions, LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS); + + cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + Assert.assertEquals( + TSDataType.INT32, + cache + .getTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME) + .getColumnSchema("s1") + .getDataType()); + } finally { + cache.invalid(TABLE_CACHE_TEST_DATABASE); + } + } + + @Test + public void delayedRollbackDoesNotEvictCommittedAlteredSchema() { + final ITableCache cache = DataNodeTableCache.getInstance(); + cache.invalid(TABLE_CACHE_TEST_DATABASE); + try { + cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, createTable(TABLE_NAME), null); + cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + + final TsTable alteredTable = createTable(TABLE_NAME); + ((FieldColumnSchema) alteredTable.getColumnSchema("s1")).setDataType(TSDataType.DOUBLE); + cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, alteredTable, null); + cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + + cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + Assert.assertEquals( + TSDataType.DOUBLE, + cache + .getTableSnapshot() + .get(TABLE_CACHE_TEST_DATABASE_NAME) + .get(TABLE_NAME) + .getColumnSchema("s1") + .getDataType()); + } finally { + cache.invalid(TABLE_CACHE_TEST_DATABASE); + } + } + + @Test + public void delayedRollbackDoesNotRestoreCommittedDeletedTable() { + final ITableCache cache = DataNodeTableCache.getInstance(); + cache.invalid(TABLE_CACHE_TEST_DATABASE); + try { + cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, createTable(TABLE_NAME), null); + cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + + cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, new PreDeleteTsTable(TABLE_NAME), null); + cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + + cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + Assert.assertFalse( + cache + .getTableSnapshot() + .getOrDefault(TABLE_CACHE_TEST_DATABASE_NAME, Collections.emptyMap()) + .containsKey(TABLE_NAME)); } finally { cache.invalid(TABLE_CACHE_TEST_DATABASE); } @@ -119,6 +215,37 @@ public void rollbackRenameTableRestoresOldName() { } } + @Test + public void rollbackAfterRestartEvictsUnknownPreviousSchema() { + final ITableCache cache = DataNodeTableCache.getInstance(); + cache.invalid(TABLE_CACHE_TEST_DATABASE); + try { + final TsTable alteredTable = createTable(TABLE_NAME); + ((FieldColumnSchema) alteredTable.getColumnSchema("s1")).setDataType(TSDataType.DOUBLE); + final byte[] initializationBytes = + TsTableInternalRPCUtil.serializeTableInitializationInfo( + Collections.singletonMap( + TABLE_CACHE_TEST_DATABASE, Collections.singletonList(alteredTable)), + Collections.singletonMap( + TABLE_CACHE_TEST_DATABASE, + Collections.singletonList(new NonCommittableTsTable(TABLE_NAME)))); + cache.init(initializationBytes); + + // A restart cannot retain the in-memory pre-update snapshot. Evict the potentially stale + // schema even if the recovered procedure repeats PRE_UPDATE, and fail closed until the + // canonical schema can be fetched from the CN. + cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, alteredTable, null); + cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + Assert.assertFalse( + cache + .getTableSnapshot() + .getOrDefault(TABLE_CACHE_TEST_DATABASE_NAME, Collections.emptyMap()) + .containsKey(TABLE_NAME)); + } finally { + cache.invalid(TABLE_CACHE_TEST_DATABASE); + } + } + private Semaphore getFetchTableSemaphore(final ITableCache cache) throws Exception { final Field field = DataNodeTableCache.class.getDeclaredField("fetchTableSemaphore"); field.setAccessible(true); diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java index b7bf58fd8bd3..806581d0e09e 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -339,4 +339,6 @@ private CommonMessages() {} "Table '%s.%s' is being deleted. Please wait for deletion to finish, or retry DROP TABLE if it is stuck."; public static final String EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROPPING_THE_COLUMN_IF_IT_IS_STUCK_875DAFFE = "Column '%s' in table '%s.%s' is being deleted. Please wait for deletion to finish, or retry dropping the column if it is stuck."; + public static final String EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_ALTERED_PLEASE_WAIT_FOR_ALTERATION_TO_FINISH_OR_RETRY_ALTERING_THE_COLUMN_IF_IT_IS_STUCK_11155B55 = + "Column '%s' in table '%s.%s' is being altered. Please wait for alteration to finish, or retry altering the column if it is stuck."; } diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index 45a682efe376..76ce8769aa13 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -236,4 +236,6 @@ private CommonMessages() {} "表 '%s.%s' 正在删除中。请等待删除完成；如果删除一直未完成，请重试 DROP TABLE。"; public static final String EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROPPING_THE_COLUMN_IF_IT_IS_STUCK_875DAFFE = "列 '%s'（位于表 '%s.%s'）正在删除中。请等待删除完成；如果删除一直未完成，请重试删除该列。"; + public static final String EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_ALTERED_PLEASE_WAIT_FOR_ALTERATION_TO_FINISH_OR_RETRY_ALTERING_THE_COLUMN_IF_IT_IS_STUCK_11155B55 = + "列 '%s'（位于表 '%s.%s'）正在修改中。请等待修改完成；如果修改一直未完成，请重试修改该列。"; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInAlterException.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInAlterException.java new file mode 100644 index 000000000000..20f24abe0627 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInAlterException.java @@ -0,0 +1,40 @@ +/* + * 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.commons.exception.table; + +import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.i18n.CommonMessages; +import org.apache.iotdb.commons.utils.PathUtils; +import org.apache.iotdb.rpc.TSStatusCode; + +public class ColumnInAlterException extends MetadataException { + + public ColumnInAlterException( + final String database, final String tableName, final String columnName) { + super( + String.format( + CommonMessages + .EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_ALTERED_PLEASE_WAIT_FOR_ALTERATION_TO_FINISH_OR_RETRY_ALTERING_THE_COLUMN_IF_IT_IS_STUCK_11155B55, + columnName, + PathUtils.unQualifyDatabaseName(database), + tableName), + TSStatusCode.SEMANTIC_ERROR.getStatusCode()); + } +}