You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2024/11/07 10:03:24 UTC
(iotdb) branch master updated: Implement delete device && Removed some unnecessary executions in multi replicasets by schema procedures && Fixed some minor bugs related to dualKeyCache
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 38f6d0f7cec Implement delete device && Removed some unnecessary executions in multi replicasets by schema procedures && Fixed some minor bugs related to dualKeyCache
38f6d0f7cec is described below
commit 38f6d0f7cecc16ebbddceb2ac388d8328e32c0fc
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Thu Nov 7 18:03:18 2024 +0800
Implement delete device && Removed some unnecessary executions in multi replicasets by schema procedures && Fixed some minor bugs related to dualKeyCache
---
.../iotdb/relational/it/schema/IoTDBDeviceIT.java | 10 +
.../java/org/apache/iotdb/jdbc/IoTDBStatement.java | 8 +-
.../client/async/CnToDnAsyncRequestType.java | 5 +
.../CnToDnInternalServiceAsyncRequestManager.java | 30 ++
.../iotdb/confignode/manager/ConfigManager.java | 10 +
.../apache/iotdb/confignode/manager/IManager.java | 4 +
.../iotdb/confignode/manager/ProcedureManager.java | 47 +++
.../impl/schema/DeactivateTemplateProcedure.java | 6 +-
.../impl/schema/DeleteTimeSeriesProcedure.java | 2 +-
.../table/AbstractAlterOrDropTableProcedure.java | 67 ++++
.../impl/schema/table/AddTableColumnProcedure.java | 12 +
.../impl/schema/table/DeleteDevicesProcedure.java | 367 +++++++++++++++++++++
.../schema/table/DropTableColumnProcedure.java | 1 +
.../impl/schema/table/DropTableProcedure.java | 72 +---
.../state/schema/DeleteDevicesState.java} | 13 +-
.../procedure/store/ProcedureFactory.java | 6 +
.../confignode/procedure/store/ProcedureType.java | 1 +
.../thrift/ConfigNodeRPCServiceProcessor.java | 7 +
.../schema/table/DeleteDevicesProcedureTest.java | 54 +++
.../schemaregion/SchemaExecutionVisitor.java | 39 +++
.../aggregate/AbstractFormalProcessor.java | 9 +-
.../iotdb/db/protocol/client/ConfigNodeClient.java | 8 +
.../impl/DataNodeInternalRPCServiceImpl.java | 188 ++++++++---
.../schema/source/DeviceAttributeUpdater.java | 30 +-
.../schema/source/DeviceBlackListConstructor.java | 85 +++++
.../schema/source/DevicePredicateHandler.java | 10 +-
.../operator/schema/source/DeviceUpdater.java | 80 +++++
.../iotdb/db/queryengine/plan/Coordinator.java | 18 +-
.../cache/schema/dualkeycache/IDualKeyCache.java | 3 +
.../schema/dualkeycache/impl/DualKeyCacheImpl.java | 52 ++-
.../execution/config/TableConfigTaskVisitor.java | 19 ++
.../config/executor/ClusterConfigTaskExecutor.java | 72 +++-
.../config/executor/IConfigTaskExecutor.java | 5 +
.../relational/AbstractAlterOrDropTableTask.java | 8 +-
.../metadata/relational/DeleteDeviceTask.java | 72 ++++
.../plan/planner/TableOperatorGenerator.java | 1 -
.../plan/planner/plan/node/PlanNodeType.java | 12 +
.../plan/planner/plan/node/PlanVisitor.java | 18 +
.../relational/analyzer/StatementAnalyzer.java | 24 +-
.../analyzer/StatementAnalyzerFactory.java | 12 +-
.../fetcher/cache/TableDeviceSchemaCache.java | 55 +++
.../plan/relational/planner/TableModelPlanner.java | 50 ++-
.../schema/AbstractTableDevicesDeletionNode.java | 91 +++++
.../schema/ConstructTableDevicesBlackListNode.java | 104 ++++++
.../node/schema/CreateOrUpdateTableDeviceNode.java | 2 +-
.../schema/DeleteTableDevicesInBlackListNode.java | 76 +++++
.../schema/RollbackTableDevicesBlackListNode.java | 76 +++++
.../relational/sql/ast/AbstractTraverseDevice.java | 4 +-
.../plan/relational/sql/ast/AstVisitor.java | 16 +-
.../plan/relational/sql/ast/CreateTable.java | 14 -
.../plan/relational/sql/ast/DeleteDevice.java | 198 +++++++++++
.../plan/relational/sql/parser/AstBuilder.java | 12 +
.../rescon/MemSchemaRegionStatistics.java | 4 +
.../schemaengine/schemaregion/ISchemaRegion.java | 15 +
.../schemaregion/ISchemaRegionPlan.java | 2 +-
.../schemaregion/SchemaRegionPlanType.java | 3 +
.../schemaregion/SchemaRegionPlanVisitor.java | 19 ++
.../update/DeviceAttributeCacheUpdater.java | 29 ++
.../attribute/update/UpdateClearContainer.java | 15 +
.../attribute/update/UpdateContainer.java | 4 +
.../attribute/update/UpdateDetailContainer.java | 73 +++-
.../schemaregion/impl/SchemaRegionMemoryImpl.java | 100 ++++++
.../schemaregion/impl/SchemaRegionPBTreeImpl.java | 21 ++
.../visitor/SchemaRegionPlanDeserializer.java | 26 +-
.../visitor/SchemaRegionPlanSerializer.java | 24 ++
.../mtree/impl/mem/MTreeBelowSGMemoryImpl.java | 120 ++++++-
.../impl/mem/mnode/factory/MemMNodeFactory.java | 6 +-
.../impl/mem/mnode/info/DatabaseDeviceInfo.java | 2 +-
.../mtree/impl/mem/mnode/info/TableDeviceInfo.java | 38 ++-
.../info/{DeviceInfo.java => TreeDeviceInfo.java} | 27 +-
.../mnode/iterator/AbstractTraverserIterator.java | 2 +-
.../mtree/impl/pbtree/MTreeBelowSGCachedImpl.java | 9 +-
.../pbtree/mnode/factory/CacheMNodeFactory.java | 6 +-
.../schemaregion/mtree/traverser/Traverser.java | 2 +-
.../schemaregion/utils/MNodeUtils.java | 4 +-
.../write/req/SchemaRegionWritePlanFactory.java | 11 +-
.../apache/iotdb/commons/conf/IoTDBConstant.java | 1 +
.../schema/node/common/DeviceMNodeWrapper.java | 16 +-
.../commons/schema/node/info/IDeviceInfo.java | 6 +-
.../commons/schema/node/role/IDeviceMNode.java | 6 +-
.../db/relational/grammar/sql/RelationalSql.g4 | 5 +-
.../src/main/thrift/confignode.thrift | 15 +
.../src/main/thrift/datanode.thrift | 46 ++-
83 files changed, 2514 insertions(+), 328 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDeviceIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDeviceIT.java
index fdf14ca38be..32f061c1f3e 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDeviceIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDeviceIT.java
@@ -229,6 +229,16 @@ public class IoTDBDeviceIT {
"insert into table0(region_id, plant_id, device_id, model, temperature, humidity) values('2', '5', '3', 'A', 37.6, 111.1)");
TestUtils.assertResultSetSize(
statement.executeQuery("show devices from table0 offset 1 limit 1"), 1);
+
+ // Test delete devices
+ statement.execute(
+ "delete devices from table0 where substring(region_id, 1, 1) in ('1', '3') and 1 + 1 = 2");
+ TestUtils.assertResultSetSize(statement.executeQuery("show devices from table0"), 1);
+
+ // Test successfully invalidate cache
+ statement.execute(
+ "insert into table0(region_id, plant_id, device_id, model, temperature, humidity) values('1', '5', '3', 'A', 37.6, 111.1)");
+ TestUtils.assertResultSetSize(statement.executeQuery("show devices from table0"), 2);
}
}
}
diff --git a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java
index c38da4dc655..def4daf643b 100644
--- a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java
+++ b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java
@@ -545,15 +545,15 @@ public class IoTDBStatement implements Statement {
throw new SQLException(NOT_SUPPORT_EXECUTE_UPDATE);
}
- private int executeUpdateSQL(String sql) throws TException, IoTDBSQLException {
- TSExecuteStatementReq execReq = new TSExecuteStatementReq(sessionId, sql, stmtId);
- TSExecuteStatementResp execResp = client.executeUpdateStatement(execReq);
+ private int executeUpdateSQL(final String sql) throws TException, IoTDBSQLException {
+ final TSExecuteStatementReq execReq = new TSExecuteStatementReq(sessionId, sql, stmtId);
+ final TSExecuteStatementResp execResp = client.executeUpdateStatement(execReq);
if (execResp.isSetQueryId()) {
queryId = execResp.getQueryId();
}
try {
RpcUtils.verifySuccess(execResp.getStatus());
- } catch (StatementExecutionException e) {
+ } catch (final StatementExecutionException e) {
throw new IoTDBSQLException(e.getMessage(), execResp.getStatus());
}
return 0;
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java
index 315a9a463b8..1b37fc82584 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java
@@ -113,4 +113,9 @@ public enum CnToDnAsyncRequestType {
INVALIDATE_TABLE_CACHE,
DELETE_DATA_FOR_DROP_TABLE,
DELETE_DEVICES_FOR_DROP_TABLE,
+ CONSTRUCT_TABLE_DEVICE_BLACK_LIST,
+ ROLLBACK_TABLE_DEVICE_BLACK_LIST,
+ INVALIDATE_MATCHED_TABLE_DEVICE_CACHE,
+ DELETE_DATA_FOR_TABLE_DEVICE,
+ DELETE_TABLE_DEVICE_IN_BLACK_LIST,
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java
index 8562f75f85a..07550882baa 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java
@@ -88,6 +88,9 @@ import org.apache.iotdb.mpp.rpc.thrift.TResetPeerListReq;
import org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListReq;
import org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListWithTemplateReq;
import org.apache.iotdb.mpp.rpc.thrift.TRollbackViewSchemaBlackListReq;
+import org.apache.iotdb.mpp.rpc.thrift.TTableDeviceDeletionWithPatternAndFilterReq;
+import org.apache.iotdb.mpp.rpc.thrift.TTableDeviceDeletionWithPatternReq;
+import org.apache.iotdb.mpp.rpc.thrift.TTableDeviceInvalidateCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTriggerLocationReq;
@@ -392,6 +395,33 @@ public class CnToDnInternalServiceAsyncRequestManager
(req, client, handler) ->
client.deleteDevicesForDropTable(
(TDeleteDataOrDevicesForDropTableReq) req, (DataNodeTSStatusRPCHandler) handler));
+ actionMapBuilder.put(
+ CnToDnAsyncRequestType.CONSTRUCT_TABLE_DEVICE_BLACK_LIST,
+ (req, client, handler) ->
+ client.constructTableDeviceBlackList(
+ (TTableDeviceDeletionWithPatternAndFilterReq) req,
+ (DataNodeTSStatusRPCHandler) handler));
+ actionMapBuilder.put(
+ CnToDnAsyncRequestType.ROLLBACK_TABLE_DEVICE_BLACK_LIST,
+ (req, client, handler) ->
+ client.rollbackTableDeviceBlackList(
+ (TTableDeviceDeletionWithPatternReq) req, (DataNodeTSStatusRPCHandler) handler));
+ actionMapBuilder.put(
+ CnToDnAsyncRequestType.INVALIDATE_MATCHED_TABLE_DEVICE_CACHE,
+ (req, client, handler) ->
+ client.invalidateMatchedTableDeviceCache(
+ (TTableDeviceInvalidateCacheReq) req, (DataNodeTSStatusRPCHandler) handler));
+ actionMapBuilder.put(
+ CnToDnAsyncRequestType.DELETE_DATA_FOR_TABLE_DEVICE,
+ (req, client, handler) ->
+ client.deleteDataForTableDevice(
+ (TTableDeviceDeletionWithPatternAndFilterReq) req,
+ (DataNodeTSStatusRPCHandler) handler));
+ actionMapBuilder.put(
+ CnToDnAsyncRequestType.DELETE_TABLE_DEVICE_IN_BLACK_LIST,
+ (req, client, handler) ->
+ client.deleteTableDeviceInBlackList(
+ (TTableDeviceDeletionWithPatternReq) req, (DataNodeTSStatusRPCHandler) handler));
actionMapBuilder.put(
CnToDnAsyncRequestType.CLEAN_DATA_NODE_CACHE,
(req, client, handler) ->
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index 1cbf4d33dbd..20dba0bfee1 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -159,6 +159,8 @@ import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.confignode.rpc.thrift.TDeactivateSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabasesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceResp;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropModelReq;
@@ -2597,6 +2599,14 @@ public class ConfigManager implements IManager {
}
}
+ @Override
+ public TDeleteTableDeviceResp deleteDevice(final TDeleteTableDeviceReq req) {
+ final TSStatus status = confirmLeader();
+ return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ ? procedureManager.deleteDevices(req)
+ : new TDeleteTableDeviceResp(status);
+ }
+
@Override
public TShowTableResp showTables(final String database, final boolean isDetails) {
final TSStatus status = confirmLeader();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
index f04691e93fe..89739d2e6c7 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
@@ -84,6 +84,8 @@ import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TDeactivateSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabasesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceResp;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropModelReq;
@@ -836,6 +838,8 @@ public interface IManager {
TSStatus alterOrDropTable(final TAlterOrDropTableReq req);
+ TDeleteTableDeviceResp deleteDevice(final TDeleteTableDeviceReq req);
+
TShowTableResp showTables(final String database, final boolean isDetails);
TFetchTableResp fetchTables(final Map<String, Set<String>> fetchTableMap);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
index df5a905cf57..9ccf4eb4154 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
@@ -88,6 +88,7 @@ import org.apache.iotdb.confignode.procedure.impl.schema.UnsetTemplateProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.AbstractAlterOrDropTableProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.AddTableColumnProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.CreateTableProcedure;
+import org.apache.iotdb.confignode.procedure.impl.schema.table.DeleteDevicesProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.DropTableColumnProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.DropTableProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.RenameTableColumnProcedure;
@@ -122,6 +123,8 @@ import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreateTopicReq;
import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceResp;
import org.apache.iotdb.confignode.rpc.thrift.TDropPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TMigrateRegionReq;
import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq;
@@ -1559,6 +1562,49 @@ public class ProcedureManager {
new DropTableProcedure(req.database, req.tableName, req.queryId));
}
+ public TDeleteTableDeviceResp deleteDevices(final TDeleteTableDeviceReq req) {
+ long procedureId;
+ DeleteDevicesProcedure procedure = null;
+ synchronized (this) {
+ final Pair<Long, Boolean> procedureIdDuplicatePair =
+ checkDuplicateTableTask(
+ req.database,
+ null,
+ req.tableName,
+ req.queryId,
+ ProcedureType.DELETE_DEVICES_PROCEDURE);
+ procedureId = procedureIdDuplicatePair.getLeft();
+
+ if (procedureId == -1) {
+ if (Boolean.TRUE.equals(procedureIdDuplicatePair.getRight())) {
+ return new TDeleteTableDeviceResp(
+ RpcUtils.getStatus(
+ TSStatusCode.OVERLAP_WITH_EXISTING_TASK,
+ "Some other task is operating table with same name."));
+ }
+ procedure =
+ new DeleteDevicesProcedure(
+ req.database,
+ req.tableName,
+ req.queryId,
+ req.getPatternInfo(),
+ req.getFilterInfo());
+ procedureId = this.executor.submitProcedure(procedure);
+ }
+ }
+ executor.getResultOrProcedure(procedureId);
+ final List<TSStatus> procedureStatus = new ArrayList<>();
+ if (waitingProcedureFinished(Collections.singletonList(procedureId), procedureStatus)) {
+ if (Objects.isNull(procedure)) {
+ procedure = ((DeleteDevicesProcedure) executor.getResultOrProcedure(procedureId));
+ }
+ return new TDeleteTableDeviceResp(StatusUtils.OK)
+ .setDeletedNum(Objects.nonNull(procedure) ? procedure.getDeletedDevicesNum() : -1);
+ } else {
+ return new TDeleteTableDeviceResp(procedureStatus.get(0));
+ }
+ }
+
private TSStatus executeWithoutDuplicate(
final String database,
final TsTable table,
@@ -1619,6 +1665,7 @@ public class ProcedureManager {
case RENAME_TABLE_COLUMN_PROCEDURE:
case DROP_TABLE_COLUMN_PROCEDURE:
case DROP_TABLE_PROCEDURE:
+ case DELETE_DEVICES_PROCEDURE:
final AbstractAlterOrDropTableProcedure<?> alterTableProcedure =
(AbstractAlterOrDropTableProcedure<?>) procedure;
if (type == thisType && queryId.equals(alterTableProcedure.getQueryId())) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
index 6f171174a61..ba987a454f3 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
@@ -193,8 +193,8 @@ public class DeactivateTemplateProcedure
return preDeletedNum;
}
- private void invalidateCache(ConfigNodeProcedureEnv env) {
- // if no target timeseres, return directly
+ private void invalidateCache(final ConfigNodeProcedureEnv env) {
+ // if no target timeseries, return directly
if (!timeSeriesPatternTree.isEmpty()) {
Map<Integer, TDataNodeLocation> dataNodeLocationMap =
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
@@ -232,7 +232,7 @@ public class DeactivateTemplateProcedure
"delete data",
env,
relatedDataRegionGroup,
- true,
+ false,
CnToDnAsyncRequestType.DELETE_DATA_FOR_DELETE_SCHEMA,
((dataNodeLocation, consensusGroupIdList) ->
new TDeleteDataForDeleteSchemaReq(
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteTimeSeriesProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteTimeSeriesProcedure.java
index 57e0921feca..2ea3640877f 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteTimeSeriesProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteTimeSeriesProcedure.java
@@ -247,7 +247,7 @@ public class DeleteTimeSeriesProcedure
"delete data",
env,
relatedDataRegionGroup,
- true,
+ false,
CnToDnAsyncRequestType.DELETE_DATA_FOR_DELETE_SCHEMA,
((dataNodeLocation, consensusGroupIdList) ->
new TDeleteDataForDeleteSchemaReq(
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractAlterOrDropTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractAlterOrDropTableProcedure.java
index 7b8ee517e85..1311aaccd39 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractAlterOrDropTableProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractAlterOrDropTableProcedure.java
@@ -19,13 +19,19 @@
package org.apache.iotdb.confignode.procedure.impl.schema.table;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
+import org.apache.iotdb.confignode.procedure.impl.schema.DataNodeRegionTaskExecutor;
import org.apache.iotdb.confignode.procedure.impl.schema.SchemaUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
@@ -34,8 +40,12 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiFunction;
public abstract class AbstractAlterOrDropTableProcedure<T>
extends StateMachineProcedure<ConfigNodeProcedureEnv, T> {
@@ -155,6 +165,63 @@ public abstract class AbstractAlterOrDropTableProcedure<T>
}
}
+ protected class TableRegionTaskExecutor<Q> extends DataNodeRegionTaskExecutor<Q, TSStatus> {
+
+ private final String taskName;
+
+ protected TableRegionTaskExecutor(
+ final String taskName,
+ final ConfigNodeProcedureEnv env,
+ final Map<TConsensusGroupId, TRegionReplicaSet> targetRegionGroup,
+ final CnToDnAsyncRequestType dataNodeRequestType,
+ final BiFunction<TDataNodeLocation, List<TConsensusGroupId>, Q> dataNodeRequestGenerator) {
+ super(env, targetRegionGroup, false, dataNodeRequestType, dataNodeRequestGenerator);
+ this.taskName = taskName;
+ }
+
+ @Override
+ protected List<TConsensusGroupId> processResponseOfOneDataNode(
+ final TDataNodeLocation dataNodeLocation,
+ final List<TConsensusGroupId> consensusGroupIdList,
+ final TSStatus response) {
+ final List<TConsensusGroupId> failedRegionList = new ArrayList<>();
+ if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return failedRegionList;
+ }
+
+ if (response.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
+ final List<TSStatus> subStatus = response.getSubStatus();
+ for (int i = 0; i < subStatus.size(); i++) {
+ if (subStatus.get(i).getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ failedRegionList.add(consensusGroupIdList.get(i));
+ }
+ }
+ } else {
+ failedRegionList.addAll(consensusGroupIdList);
+ }
+ return failedRegionList;
+ }
+
+ @Override
+ protected void onAllReplicasetFailure(
+ final TConsensusGroupId consensusGroupId,
+ final Set<TDataNodeLocation> dataNodeLocationSet) {
+ setFailure(
+ new ProcedureException(
+ new MetadataException(
+ String.format(
+ "[%s] for %s.%s failed when [%s] because failed to execute in all replicaset of %s %s. Failure nodes: %s",
+ this.getClass().getSimpleName(),
+ database,
+ tableName,
+ taskName,
+ consensusGroupId.type,
+ consensusGroupId.id,
+ dataNodeLocationSet))));
+ interruptTask();
+ }
+ }
+
@Override
public boolean equals(final Object o) {
if (this == o) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java
index 5a710b001cc..d634e5efefe 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java
@@ -40,6 +40,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.Objects;
public class AddTableColumnProcedure
extends AbstractAlterOrDropTableProcedure<AddTableColumnState> {
@@ -201,4 +202,15 @@ public class AddTableColumnProcedure
this.addedColumnList = TsTableColumnSchemaUtil.deserializeColumnSchemaList(byteBuffer);
}
+
+ @Override
+ public boolean equals(final Object o) {
+ return super.equals(o)
+ && Objects.equals(this.addedColumnList, ((AddTableColumnProcedure) o).addedColumnList);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), addedColumnList);
+ }
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DeleteDevicesProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DeleteDevicesProcedure.java
new file mode 100644
index 00000000000..11dd5b1187c
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DeleteDevicesProcedure.java
@@ -0,0 +1,367 @@
+/*
+ * 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.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+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.path.PartialPath;
+import org.apache.iotdb.commons.path.PathPatternTree;
+import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
+import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager;
+import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
+import org.apache.iotdb.confignode.procedure.impl.schema.DataNodeRegionTaskExecutor;
+import org.apache.iotdb.confignode.procedure.state.schema.DeleteDevicesState;
+import org.apache.iotdb.confignode.procedure.store.ProcedureType;
+import org.apache.iotdb.mpp.rpc.thrift.TTableDeviceDeletionWithPatternAndFilterReq;
+import org.apache.iotdb.mpp.rpc.thrift.TTableDeviceDeletionWithPatternReq;
+import org.apache.iotdb.mpp.rpc.thrift.TTableDeviceInvalidateCacheReq;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD;
+import static org.apache.iotdb.commons.schema.SchemaConstant.ROOT;
+import static org.apache.iotdb.confignode.procedure.state.schema.DeleteDevicesState.CLEAN_DATANODE_SCHEMA_CACHE;
+import static org.apache.iotdb.confignode.procedure.state.schema.DeleteDevicesState.CONSTRUCT_BLACK_LIST;
+import static org.apache.iotdb.confignode.procedure.state.schema.DeleteDevicesState.DELETE_DATA;
+import static org.apache.iotdb.confignode.procedure.state.schema.DeleteDevicesState.DELETE_DEVICE_SCHEMA;
+import static org.apache.iotdb.rpc.TSStatusCode.TABLE_ALREADY_EXISTS;
+
+public class DeleteDevicesProcedure extends AbstractAlterOrDropTableProcedure<DeleteDevicesState> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(DeleteDevicesProcedure.class);
+ private byte[] patternBytes;
+ private byte[] filterBytes;
+
+ // Transient
+ private PathPatternTree patternTree;
+
+ // Transient, will not be returned if once recovers
+ private long deletedDevicesNum;
+
+ public DeleteDevicesProcedure() {
+ super();
+ }
+
+ public DeleteDevicesProcedure(
+ final String database,
+ final String tableName,
+ final String queryId,
+ final byte[] patternBytes,
+ final byte[] filterBytes) {
+ super(database, tableName, queryId);
+ this.patternBytes = patternBytes;
+ this.filterBytes = filterBytes;
+ }
+
+ @Override
+ protected Flow executeFromState(final ConfigNodeProcedureEnv env, final DeleteDevicesState state)
+ throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException {
+ final long startTime = System.currentTimeMillis();
+ try {
+ switch (state) {
+ case CHECK_TABLE_EXISTENCE:
+ LOGGER.info("Check the existence of table {}.{}", database, table.getTableName());
+ checkTableExistence(env);
+ break;
+ case CONSTRUCT_BLACK_LIST:
+ LOGGER.info("Construct schemaEngine black list of devices in {}.{}", database, tableName);
+ constructBlackList(env);
+ if (deletedDevicesNum > 0) {
+ setNextState(CLEAN_DATANODE_SCHEMA_CACHE);
+ break;
+ } else {
+ return Flow.NO_MORE_STATE;
+ }
+ case CLEAN_DATANODE_SCHEMA_CACHE:
+ LOGGER.info("Invalidate cache of devices in {}.{}", database, tableName);
+ invalidateCache(env);
+ break;
+ case DELETE_DATA:
+ LOGGER.info("Delete data of devices in {}.{}", database, tableName);
+ deleteData(env);
+ break;
+ case DELETE_DEVICE_SCHEMA:
+ LOGGER.info("Delete devices in {}.{} in schemaEngine", database, tableName);
+ deleteDeviceSchema(env);
+ return Flow.NO_MORE_STATE;
+ default:
+ setFailure(new ProcedureException("Unrecognized state " + state));
+ return Flow.NO_MORE_STATE;
+ }
+ return Flow.HAS_MORE_STATE;
+ } finally {
+ LOGGER.info(
+ "DeleteTimeSeries-[{}] costs {}ms", state, (System.currentTimeMillis() - startTime));
+ }
+ }
+
+ private void checkTableExistence(final ConfigNodeProcedureEnv env) {
+ if (env.getConfigManager()
+ .getClusterSchemaManager()
+ .getTableIfExists(database, table.getTableName())
+ .isPresent()) {
+ setFailure(
+ new ProcedureException(
+ new IoTDBException(
+ String.format(
+ "Table '%s.%s' already exists.",
+ database.substring(ROOT.length() + 1), table.getTableName()),
+ TABLE_ALREADY_EXISTS.getStatusCode())));
+ } else {
+ setNextState(CONSTRUCT_BLACK_LIST);
+ }
+ }
+
+ private void constructBlackList(final ConfigNodeProcedureEnv env) {
+ patternTree = new PathPatternTree();
+ final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ final DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ final PartialPath path;
+ try {
+ path = new PartialPath(new String[] {ROOT, database.substring(5), tableName});
+ patternTree.appendPathPattern(path);
+ patternTree.appendPathPattern(path.concatAsMeasurementPath(MULTI_LEVEL_PATH_WILDCARD));
+ patternTree.serialize(dataOutputStream);
+ } catch (final IOException e) {
+ LOGGER.warn("failed to serialize request for table {}.{}", database, table.getTableName(), e);
+ }
+
+ final Map<TConsensusGroupId, TRegionReplicaSet> relatedSchemaRegionGroup =
+ env.getConfigManager().getRelatedSchemaRegionGroup(patternTree, true);
+
+ if (relatedSchemaRegionGroup.isEmpty()) {
+ deletedDevicesNum = 0;
+ return;
+ }
+ final List<TSStatus> successResult = new ArrayList<>();
+ new DataNodeRegionTaskExecutor<TTableDeviceDeletionWithPatternAndFilterReq, TSStatus>(
+ env,
+ relatedSchemaRegionGroup,
+ false,
+ CnToDnAsyncRequestType.CONSTRUCT_TABLE_DEVICE_BLACK_LIST,
+ ((dataNodeLocation, consensusGroupIdList) ->
+ new TTableDeviceDeletionWithPatternAndFilterReq(
+ new ArrayList<>(consensusGroupIdList),
+ tableName,
+ ByteBuffer.wrap(patternBytes),
+ ByteBuffer.wrap(filterBytes)))) {
+ @Override
+ protected List<TConsensusGroupId> processResponseOfOneDataNode(
+ final TDataNodeLocation dataNodeLocation,
+ final List<TConsensusGroupId> consensusGroupIdList,
+ final TSStatus response) {
+ final List<TConsensusGroupId> failedRegionList = new ArrayList<>();
+ if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ successResult.add(response);
+ } else if (response.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
+ final List<TSStatus> subStatusList = response.getSubStatus();
+ for (int i = 0; i < subStatusList.size(); i++) {
+ if (subStatusList.get(i).getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ successResult.add(subStatusList.get(i));
+ } else {
+ failedRegionList.add(consensusGroupIdList.get(i));
+ }
+ }
+ } else {
+ failedRegionList.addAll(consensusGroupIdList);
+ }
+ return failedRegionList;
+ }
+
+ @Override
+ protected void onAllReplicasetFailure(
+ final TConsensusGroupId consensusGroupId,
+ final Set<TDataNodeLocation> dataNodeLocationSet) {
+ setFailure(
+ new ProcedureException(
+ new MetadataException(
+ String.format(
+ "[%s] for %s.%s failed when construct black list for table because failed to execute in all replicaset of %s %s. Failure nodes: %s",
+ this.getClass().getSimpleName(),
+ database,
+ tableName,
+ consensusGroupId.type,
+ consensusGroupId.id,
+ dataNodeLocationSet))));
+ interruptTask();
+ }
+ }.execute();
+
+ setNextState(CONSTRUCT_BLACK_LIST);
+ deletedDevicesNum =
+ !isFailed()
+ ? successResult.stream()
+ .mapToLong(resp -> Long.parseLong(resp.getMessage()))
+ .reduce(Long::sum)
+ .orElse(0L)
+ : 0;
+ }
+
+ private void invalidateCache(final ConfigNodeProcedureEnv env) {
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ final DataNodeAsyncRequestContext<TTableDeviceInvalidateCacheReq, TSStatus> clientHandler =
+ new DataNodeAsyncRequestContext<>(
+ CnToDnAsyncRequestType.INVALIDATE_MATCHED_TABLE_DEVICE_CACHE,
+ new TTableDeviceInvalidateCacheReq(database, tableName, ByteBuffer.wrap(patternBytes)),
+ dataNodeLocationMap);
+ CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
+ final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (final TSStatus status : statusMap.values()) {
+ // All dataNodes must clear the related schemaEngine cache
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.error(
+ "Failed to invalidate schemaEngine cache of devices in table {}.{}",
+ database,
+ tableName);
+ setFailure(
+ new ProcedureException(new MetadataException("Invalidate schemaEngine cache failed")));
+ return;
+ }
+ }
+
+ setNextState(DELETE_DATA);
+ }
+
+ private void deleteData(final ConfigNodeProcedureEnv env) {
+ new TableRegionTaskExecutor<>(
+ "delete data for table device",
+ env,
+ env.getConfigManager().getRelatedDataRegionGroup(patternTree, true),
+ CnToDnAsyncRequestType.DELETE_DATA_FOR_TABLE_DEVICE,
+ (dataNodeLocation, consensusGroupIdList) ->
+ new TTableDeviceDeletionWithPatternAndFilterReq(
+ consensusGroupIdList,
+ tableName,
+ ByteBuffer.wrap(patternBytes),
+ ByteBuffer.wrap(filterBytes)))
+ .execute();
+ setNextState(DELETE_DEVICE_SCHEMA);
+ }
+
+ private void deleteDeviceSchema(final ConfigNodeProcedureEnv env) {
+ new TableRegionTaskExecutor<>(
+ "roll back table device black list",
+ env,
+ env.getConfigManager().getRelatedSchemaRegionGroup(patternTree, true),
+ CnToDnAsyncRequestType.DELETE_TABLE_DEVICE_IN_BLACK_LIST,
+ (dataNodeLocation, consensusGroupIdList) ->
+ new TTableDeviceDeletionWithPatternReq(
+ consensusGroupIdList, tableName, ByteBuffer.wrap(patternBytes)))
+ .execute();
+ }
+
+ @Override
+ protected void rollbackState(
+ final ConfigNodeProcedureEnv env, final DeleteDevicesState deleteDevicesState)
+ throws IOException, InterruptedException, ProcedureException {
+ if (deleteDevicesState == CONSTRUCT_BLACK_LIST) {
+ new TableRegionTaskExecutor<>(
+ "roll back table device black list",
+ env,
+ env.getConfigManager().getRelatedSchemaRegionGroup(patternTree, true),
+ CnToDnAsyncRequestType.ROLLBACK_TABLE_DEVICE_BLACK_LIST,
+ (dataNodeLocation, consensusGroupIdList) ->
+ new TTableDeviceDeletionWithPatternReq(
+ consensusGroupIdList, tableName, ByteBuffer.wrap(patternBytes)))
+ .execute();
+ }
+ }
+
+ public long getDeletedDevicesNum() {
+ return deletedDevicesNum;
+ }
+
+ @Override
+ protected DeleteDevicesState getState(final int stateId) {
+ return DeleteDevicesState.values()[stateId];
+ }
+
+ @Override
+ protected int getStateId(final DeleteDevicesState deleteDevicesState) {
+ return deleteDevicesState.ordinal();
+ }
+
+ @Override
+ protected DeleteDevicesState getInitialState() {
+ return CONSTRUCT_BLACK_LIST;
+ }
+
+ @Override
+ protected String getActionMessage() {
+ // Not used
+ return null;
+ }
+
+ @Override
+ public void serialize(final DataOutputStream stream) throws IOException {
+ stream.writeShort(ProcedureType.DELETE_DEVICES_PROCEDURE.getTypeCode());
+ super.serialize(stream);
+
+ ReadWriteIOUtils.write(patternBytes.length, stream);
+ stream.write(patternBytes);
+ ReadWriteIOUtils.write(filterBytes.length, stream);
+ stream.write(filterBytes);
+ }
+
+ @Override
+ public void deserialize(final ByteBuffer byteBuffer) {
+ super.deserialize(byteBuffer);
+
+ patternBytes = new byte[ReadWriteIOUtils.readInt(byteBuffer)];
+ byteBuffer.get(patternBytes);
+ filterBytes = new byte[ReadWriteIOUtils.readInt(byteBuffer)];
+ byteBuffer.get(filterBytes);
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ return super.equals(o)
+ && Arrays.equals(this.patternBytes, ((DeleteDevicesProcedure) o).patternBytes)
+ && Arrays.equals(this.filterBytes, ((DeleteDevicesProcedure) o).filterBytes);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ super.hashCode(), Arrays.hashCode(patternBytes), Arrays.hashCode(filterBytes));
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java
index 60c9386739a..559d1ad6bfb 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java
@@ -33,6 +33,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
+// TODO
public class DropTableColumnProcedure
extends AbstractAlterOrDropTableProcedure<DropTableColumnState> {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java
index d60d60861ce..20c76108fa1 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java
@@ -36,7 +36,6 @@ import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
-import org.apache.iotdb.confignode.procedure.impl.schema.DataNodeRegionTaskExecutor;
import org.apache.iotdb.confignode.procedure.impl.schema.SchemaUtils;
import org.apache.iotdb.confignode.procedure.state.schema.DropTableState;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
@@ -51,11 +50,8 @@ import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Set;
-import java.util.function.BiFunction;
import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD;
import static org.apache.iotdb.commons.schema.SchemaConstant.ROOT;
@@ -175,11 +171,10 @@ public class DropTableProcedure extends AbstractAlterOrDropTableProcedure<DropTa
env.getConfigManager().getRelatedDataRegionGroup(patternTree, true);
if (!relatedDataRegionGroup.isEmpty()) {
- new DropTableRegionTaskExecutor<>(
+ new TableRegionTaskExecutor<>(
"delete data for drop table",
env,
relatedDataRegionGroup,
- true,
CnToDnAsyncRequestType.DELETE_DATA_FOR_DROP_TABLE,
((dataNodeLocation, consensusGroupIdList) ->
new TDeleteDataOrDevicesForDropTableReq(
@@ -211,11 +206,10 @@ public class DropTableProcedure extends AbstractAlterOrDropTableProcedure<DropTa
env.getConfigManager().getRelatedSchemaRegionGroup(patternTree, true);
if (!relatedSchemaRegionGroup.isEmpty()) {
- new DropTableRegionTaskExecutor<>(
+ new TableRegionTaskExecutor<>(
"delete devices for drop table",
env,
relatedSchemaRegionGroup,
- true,
CnToDnAsyncRequestType.DELETE_DEVICES_FOR_DROP_TABLE,
((dataNodeLocation, consensusGroupIdList) ->
new TDeleteDataOrDevicesForDropTableReq(
@@ -266,66 +260,4 @@ public class DropTableProcedure extends AbstractAlterOrDropTableProcedure<DropTa
stream.writeShort(ProcedureType.DROP_TABLE_PROCEDURE.getTypeCode());
super.serialize(stream);
}
-
- private class DropTableRegionTaskExecutor<Q> extends DataNodeRegionTaskExecutor<Q, TSStatus> {
-
- private final String taskName;
-
- DropTableRegionTaskExecutor(
- final String taskName,
- final ConfigNodeProcedureEnv env,
- final Map<TConsensusGroupId, TRegionReplicaSet> targetDataRegionGroup,
- final boolean executeOnAllReplicaset,
- final CnToDnAsyncRequestType dataNodeRequestType,
- final BiFunction<TDataNodeLocation, List<TConsensusGroupId>, Q> dataNodeRequestGenerator) {
- super(
- env,
- targetDataRegionGroup,
- executeOnAllReplicaset,
- dataNodeRequestType,
- dataNodeRequestGenerator);
- this.taskName = taskName;
- }
-
- @Override
- protected List<TConsensusGroupId> processResponseOfOneDataNode(
- final TDataNodeLocation dataNodeLocation,
- final List<TConsensusGroupId> consensusGroupIdList,
- final TSStatus response) {
- final List<TConsensusGroupId> failedRegionList = new ArrayList<>();
- if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- return failedRegionList;
- }
-
- if (response.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
- final List<TSStatus> subStatus = response.getSubStatus();
- for (int i = 0; i < subStatus.size(); i++) {
- if (subStatus.get(i).getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- failedRegionList.add(consensusGroupIdList.get(i));
- }
- }
- } else {
- failedRegionList.addAll(consensusGroupIdList);
- }
- return failedRegionList;
- }
-
- @Override
- protected void onAllReplicasetFailure(
- final TConsensusGroupId consensusGroupId,
- final Set<TDataNodeLocation> dataNodeLocationSet) {
- setFailure(
- new ProcedureException(
- new MetadataException(
- String.format(
- "Drop table %s.%s failed when [%s] because failed to execute in all replicaset of %s %s. Failure nodes: %s",
- database,
- tableName,
- taskName,
- consensusGroupId.type,
- consensusGroupId.id,
- dataNodeLocationSet))));
- interruptTask();
- }
- }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegionPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/schema/DeleteDevicesState.java
similarity index 79%
copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegionPlan.java
copy to iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/schema/DeleteDevicesState.java
index f48d3902d0a..974a2fdfb54 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegionPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/schema/DeleteDevicesState.java
@@ -17,11 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.db.schemaengine.schemaregion;
+package org.apache.iotdb.confignode.procedure.state.schema;
-public interface ISchemaRegionPlan {
-
- SchemaRegionPlanType getPlanType();
-
- <R, C> R accept(SchemaRegionPlanVisitor<R, C> visitor, C context);
+public enum DeleteDevicesState {
+ CHECK_TABLE_EXISTENCE,
+ CONSTRUCT_BLACK_LIST,
+ CLEAN_DATANODE_SCHEMA_CACHE,
+ DELETE_DATA,
+ DELETE_DEVICE_SCHEMA
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
index 402d5b6ed28..7cb7f1ebeae 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
@@ -52,6 +52,7 @@ import org.apache.iotdb.confignode.procedure.impl.schema.SetTemplateProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.UnsetTemplateProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.AddTableColumnProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.CreateTableProcedure;
+import org.apache.iotdb.confignode.procedure.impl.schema.table.DeleteDevicesProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.DropTableColumnProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.DropTableProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.table.RenameTableColumnProcedure;
@@ -209,6 +210,9 @@ public class ProcedureFactory implements IProcedureFactory {
case DROP_TABLE_PROCEDURE:
procedure = new DropTableProcedure();
break;
+ case DELETE_DEVICES_PROCEDURE:
+ procedure = new DeleteDevicesProcedure();
+ break;
case CREATE_PIPE_PLUGIN_PROCEDURE:
procedure = new CreatePipePluginProcedure();
break;
@@ -369,6 +373,8 @@ public class ProcedureFactory implements IProcedureFactory {
return ProcedureType.DROP_TABLE_COLUMN_PROCEDURE;
} else if (procedure instanceof DropTableProcedure) {
return ProcedureType.DROP_TABLE_PROCEDURE;
+ } else if (procedure instanceof DeleteDevicesProcedure) {
+ return ProcedureType.DELETE_DEVICES_PROCEDURE;
} else if (procedure instanceof CreatePipePluginProcedure) {
return ProcedureType.CREATE_PIPE_PLUGIN_PROCEDURE;
} else if (procedure instanceof DropPipePluginProcedure) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
index 206cc99277c..e423452d7ad 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
@@ -68,6 +68,7 @@ public enum ProcedureType {
SET_TABLE_PROPERTIES_PROCEDURE((short) 753),
RENAME_TABLE_COLUMN_PROCEDURE((short) 754),
DROP_TABLE_COLUMN_PROCEDURE((short) 755),
+ DELETE_DEVICES_PROCEDURE((short) 756),
/** AI Model */
CREATE_MODEL_PROCEDURE((short) 800),
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index adaf19563aa..19791c2cf66 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -125,6 +125,8 @@ import org.apache.iotdb.confignode.rpc.thrift.TDeactivateSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabaseReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabasesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceResp;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq;
@@ -1314,4 +1316,9 @@ public class ConfigNodeRPCServiceProcessor implements IConfigNodeRPCService.Ifac
public TFetchTableResp fetchTables(final Map<String, Set<String>> fetchTableMap) {
return configManager.fetchTables(fetchTableMap);
}
+
+ @Override
+ public TDeleteTableDeviceResp deleteDevice(final TDeleteTableDeviceReq req) {
+ return configManager.deleteDevice(req);
+ }
}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DeleteDevicesProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DeleteDevicesProcedureTest.java
new file mode 100644
index 00000000000..ae065003a44
--- /dev/null
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DeleteDevicesProcedureTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.commons.exception.IllegalPathException;
+import org.apache.iotdb.confignode.procedure.store.ProcedureType;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class DeleteDevicesProcedureTest {
+ @Test
+ public void serializeDeserializeTest() throws IllegalPathException, IOException {
+ final DeleteDevicesProcedure deleteDevicesProcedure =
+ new DeleteDevicesProcedure(
+ "root.database1", "table1", "0", new byte[] {0, 1, 2}, new byte[] {0, 1, 2});
+
+ final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ final DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ deleteDevicesProcedure.serialize(dataOutputStream);
+
+ final ByteBuffer byteBuffer = ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+
+ Assert.assertEquals(
+ ProcedureType.DELETE_DEVICES_PROCEDURE.getTypeCode(), byteBuffer.getShort());
+
+ final DeleteDevicesProcedure deserializedProcedure = new DeleteDevicesProcedure();
+ deserializedProcedure.deserialize(byteBuffer);
+
+ Assert.assertEquals(deleteDevicesProcedure, deserializedProcedure);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaExecutionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaExecutionVisitor.java
index 3e437f72a37..51d6fccfbd4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaExecutionVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaExecutionVisitor.java
@@ -57,8 +57,11 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.vie
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedNonWritePlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedWritePlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeOperateSchemaQueueNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode;
@@ -643,6 +646,42 @@ public class SchemaExecutionVisitor extends PlanVisitor<TSStatus, ISchemaRegion>
}
}
+ @Override
+ public TSStatus visitConstructTableDevicesBlackList(
+ final ConstructTableDevicesBlackListNode node, final ISchemaRegion schemaRegion) {
+ try {
+ final long preDeletedNum = schemaRegion.constructTableDevicesBlackList(node);
+ return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, String.valueOf(preDeletedNum));
+ } catch (final MetadataException e) {
+ logger.error(e.getMessage(), e);
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ }
+ }
+
+ @Override
+ public TSStatus visitRollbackTableDevicesBlackList(
+ final RollbackTableDevicesBlackListNode node, final ISchemaRegion schemaRegion) {
+ try {
+ schemaRegion.rollbackTableDevicesBlackList(node);
+ return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ } catch (final MetadataException e) {
+ logger.error(e.getMessage(), e);
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ }
+ }
+
+ @Override
+ public TSStatus visitDeleteTableDevicesInBlackList(
+ final DeleteTableDevicesInBlackListNode node, final ISchemaRegion schemaRegion) {
+ try {
+ schemaRegion.deleteTableDevicesInBlackList(node);
+ return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ } catch (final MetadataException e) {
+ logger.error(e.getMessage(), e);
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ }
+ }
+
@Override
public TSStatus visitPipeEnrichedWritePlanNode(
final PipeEnrichedWritePlanNode node, final ISchemaRegion schemaRegion) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AbstractFormalProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AbstractFormalProcessor.java
index 357d7a0515e..b74227c16ae 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AbstractFormalProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AbstractFormalProcessor.java
@@ -34,20 +34,23 @@ import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
public abstract class AbstractFormalProcessor implements PipeProcessor {
@Override
public final void process(
- TabletInsertionEvent tabletInsertionEvent, EventCollector eventCollector) throws Exception {
+ final TabletInsertionEvent tabletInsertionEvent, final EventCollector eventCollector)
+ throws Exception {
throw new UnsupportedOperationException(
"The abstract formal processor does not support process events");
}
@Override
public final void process(
- TsFileInsertionEvent tsFileInsertionEvent, EventCollector eventCollector) throws Exception {
+ final TsFileInsertionEvent tsFileInsertionEvent, final EventCollector eventCollector)
+ throws Exception {
throw new UnsupportedOperationException(
"The abstract formal processor does not support process events");
}
@Override
- public final void process(Event event, EventCollector eventCollector) throws Exception {
+ public final void process(final Event event, final EventCollector eventCollector)
+ throws Exception {
throw new UnsupportedOperationException(
"The abstract formal processor does not support process events");
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
index 93f9bf1e77c..6e5057158ad 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
@@ -88,6 +88,8 @@ import org.apache.iotdb.confignode.rpc.thrift.TDeactivateSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabaseReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabasesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceResp;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq;
@@ -1291,6 +1293,12 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie
() -> client.fetchTables(fetchTableMap), resp -> !updateConfigNodeLeader(resp.status));
}
+ @Override
+ public TDeleteTableDeviceResp deleteDevice(final TDeleteTableDeviceReq req) throws TException {
+ return executeRemoteCallWithRetry(
+ () -> client.deleteDevice(req), resp -> !updateConfigNodeLeader(resp.status));
+ }
+
public static class Factory extends ThriftClientFactory<ConfigRegionId, ConfigNodeClient> {
public Factory(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 76583044664..202140cceb7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -141,8 +141,12 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNo
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableDeviceSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceSchemaCache;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TreeDeviceSchemaCacheManager;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableSchemaQueryWriteVisitor;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DeleteDevice;
import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler;
import org.apache.iotdb.db.queryengine.plan.statement.component.WhereCondition;
import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement;
@@ -251,6 +255,9 @@ import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeResp;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceResp;
import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeResp;
+import org.apache.iotdb.mpp.rpc.thrift.TTableDeviceDeletionWithPatternAndFilterReq;
+import org.apache.iotdb.mpp.rpc.thrift.TTableDeviceDeletionWithPatternReq;
+import org.apache.iotdb.mpp.rpc.thrift.TTableDeviceInvalidateCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TTsFilePieceReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
@@ -676,25 +683,23 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TSStatus deleteDataForDeleteSchema(TDeleteDataForDeleteSchemaReq req) {
- PathPatternTree patternTree =
- PathPatternTree.deserialize(ByteBuffer.wrap(req.getPathPatternTree()));
- List<MeasurementPath> pathList = patternTree.getAllPathPatterns(true);
+ public TSStatus deleteDataForDeleteSchema(final TDeleteDataForDeleteSchemaReq req) {
+ final List<MeasurementPath> pathList =
+ PathPatternTree.deserialize(ByteBuffer.wrap(req.getPathPatternTree()))
+ .getAllPathPatterns(true);
return executeInternalSchemaTask(
req.getDataRegionIdList(),
- consensusGroupId -> {
- RegionWriteExecutor executor = new RegionWriteExecutor();
- return executor
- .execute(
- new DataRegionId(consensusGroupId.getId()),
- req.isSetIsGeneratedByPipe() && req.isIsGeneratedByPipe()
- ? new PipeEnrichedDeleteDataNode(
- new DeleteDataNode(
- new PlanNodeId(""), pathList, Long.MIN_VALUE, Long.MAX_VALUE))
- : new DeleteDataNode(
- new PlanNodeId(""), pathList, Long.MIN_VALUE, Long.MAX_VALUE))
- .getStatus();
- });
+ consensusGroupId ->
+ new RegionWriteExecutor()
+ .execute(
+ new DataRegionId(consensusGroupId.getId()),
+ req.isSetIsGeneratedByPipe() && req.isIsGeneratedByPipe()
+ ? new PipeEnrichedDeleteDataNode(
+ new DeleteDataNode(
+ new PlanNodeId(""), pathList, Long.MIN_VALUE, Long.MAX_VALUE))
+ : new DeleteDataNode(
+ new PlanNodeId(""), pathList, Long.MIN_VALUE, Long.MAX_VALUE))
+ .getStatus());
}
@Override
@@ -913,10 +918,11 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TCheckTimeSeriesExistenceResp checkTimeSeriesExistence(TCheckTimeSeriesExistenceReq req) {
- PathPatternTree patternTree = PathPatternTree.deserialize(req.patternTree);
- TCheckTimeSeriesExistenceResp resp = new TCheckTimeSeriesExistenceResp();
- TSStatus status =
+ public TCheckTimeSeriesExistenceResp checkTimeSeriesExistence(
+ final TCheckTimeSeriesExistenceReq req) {
+ final PathPatternTree patternTree = PathPatternTree.deserialize(req.patternTree);
+ final TCheckTimeSeriesExistenceResp resp = new TCheckTimeSeriesExistenceResp();
+ final TSStatus status =
executeInternalSchemaTask(
req.getSchemaRegionIdList(),
consensusGroupId -> {
@@ -925,23 +931,23 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
// Check timeseries existence for set template shall block all timeseries creation
readWriteLock.writeLock().lock();
try {
- ISchemaRegion schemaRegion =
+ final ISchemaRegion schemaRegion =
schemaEngine.getSchemaRegion(new SchemaRegionId(consensusGroupId.getId()));
- PathPatternTree filteredPatternTree =
+ final PathPatternTree filteredPatternTree =
filterPathPatternTree(patternTree, schemaRegion.getDatabaseFullPath());
if (filteredPatternTree.isEmpty()) {
return RpcUtils.SUCCESS_STATUS;
}
- for (PartialPath pattern : filteredPatternTree.getAllPathPatterns()) {
+ for (final PartialPath pattern : filteredPatternTree.getAllPathPatterns()) {
ISchemaSource<ITimeSeriesSchemaInfo> schemaSource =
SchemaSourceFactory.getTimeSeriesSchemaCountSource(
pattern, false, null, null, SchemaConstant.ALL_MATCH_SCOPE);
- try (ISchemaReader<ITimeSeriesSchemaInfo> schemaReader =
+ try (final ISchemaReader<ITimeSeriesSchemaInfo> schemaReader =
schemaSource.getSchemaReader(schemaRegion)) {
if (schemaReader.hasNext()) {
return RpcUtils.getStatus(TSStatusCode.TIMESERIES_ALREADY_EXIST);
}
- } catch (Exception e) {
+ } catch (final Exception e) {
LOGGER.warn(e.getMessage(), e);
return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage());
}
@@ -956,7 +962,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
resp.setExists(false);
} else if (status.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
boolean hasFailure = false;
- for (TSStatus subStatus : status.getSubStatus()) {
+ for (final TSStatus subStatus : status.getSubStatus()) {
if (subStatus.getCode() == TSStatusCode.TIMESERIES_ALREADY_EXIST.getStatusCode()) {
resp.setExists(true);
} else if (subStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
@@ -976,25 +982,25 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TSStatus constructViewSchemaBlackList(TConstructViewSchemaBlackListReq req) {
- PathPatternTree patternTree =
+ public TSStatus constructViewSchemaBlackList(final TConstructViewSchemaBlackListReq req) {
+ final PathPatternTree patternTree =
PathPatternTree.deserialize(ByteBuffer.wrap(req.getPathPatternTree()));
- AtomicInteger preDeletedNum = new AtomicInteger(0);
- TSStatus executionResult =
+ final AtomicInteger preDeletedNum = new AtomicInteger(0);
+ final TSStatus executionResult =
executeInternalSchemaTask(
req.getSchemaRegionIdList(),
consensusGroupId -> {
- String storageGroup =
+ final String storageGroup =
schemaEngine
.getSchemaRegion(new SchemaRegionId(consensusGroupId.getId()))
.getDatabaseFullPath();
- PathPatternTree filteredPatternTree =
+ final PathPatternTree filteredPatternTree =
filterPathPatternTree(patternTree, storageGroup);
if (filteredPatternTree.isEmpty()) {
return RpcUtils.SUCCESS_STATUS;
}
- RegionWriteExecutor executor = new RegionWriteExecutor();
- TSStatus status =
+ final RegionWriteExecutor executor = new RegionWriteExecutor();
+ final TSStatus status =
executor
.execute(
new SchemaRegionId(consensusGroupId.getId()),
@@ -1011,21 +1017,22 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TSStatus rollbackViewSchemaBlackList(TRollbackViewSchemaBlackListReq req) {
- PathPatternTree patternTree =
+ public TSStatus rollbackViewSchemaBlackList(final TRollbackViewSchemaBlackListReq req) {
+ final PathPatternTree patternTree =
PathPatternTree.deserialize(ByteBuffer.wrap(req.getPathPatternTree()));
return executeInternalSchemaTask(
req.getSchemaRegionIdList(),
consensusGroupId -> {
- String storageGroup =
+ final String storageGroup =
schemaEngine
.getSchemaRegion(new SchemaRegionId(consensusGroupId.getId()))
.getDatabaseFullPath();
- PathPatternTree filteredPatternTree = filterPathPatternTree(patternTree, storageGroup);
+ final PathPatternTree filteredPatternTree =
+ filterPathPatternTree(patternTree, storageGroup);
if (filteredPatternTree.isEmpty()) {
return RpcUtils.SUCCESS_STATUS;
}
- RegionWriteExecutor executor = new RegionWriteExecutor();
+ final RegionWriteExecutor executor = new RegionWriteExecutor();
return executor
.execute(
new SchemaRegionId(consensusGroupId.getId()),
@@ -1096,14 +1103,14 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TPushPipeMetaResp pushPipeMeta(TPushPipeMetaReq req) {
- final List<PipeMeta> pipeMetas = new ArrayList<>();
- for (ByteBuffer byteBuffer : req.getPipeMetas()) {
- pipeMetas.add(PipeMeta.deserialize(byteBuffer));
- }
+ public TPushPipeMetaResp pushPipeMeta(final TPushPipeMetaReq req) {
try {
- List<TPushPipeMetaRespExceptionMessage> exceptionMessages =
- PipeDataNodeAgent.task().handlePipeMetaChanges(pipeMetas);
+ final List<TPushPipeMetaRespExceptionMessage> exceptionMessages =
+ PipeDataNodeAgent.task()
+ .handlePipeMetaChanges(
+ req.getPipeMetas().stream()
+ .map(PipeMeta::deserialize)
+ .collect(Collectors.toList()));
return exceptionMessages.isEmpty()
? new TPushPipeMetaResp()
@@ -1111,7 +1118,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
: new TPushPipeMetaResp()
.setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
.setExceptionMessages(exceptionMessages);
- } catch (Exception e) {
+ } catch (final Exception e) {
LOGGER.error("Error occurred when pushing pipe meta", e);
return new TPushPipeMetaResp()
.setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()));
@@ -1119,9 +1126,9 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TPushPipeMetaResp pushSinglePipeMeta(TPushSinglePipeMetaReq req) {
+ public TPushPipeMetaResp pushSinglePipeMeta(final TPushSinglePipeMetaReq req) {
try {
- TPushPipeMetaRespExceptionMessage exceptionMessage;
+ final TPushPipeMetaRespExceptionMessage exceptionMessage;
if (req.isSetPipeNameToDrop()) {
exceptionMessage = PipeDataNodeAgent.task().handleDropPipe(req.getPipeNameToDrop());
} else if (req.isSetPipeMeta()) {
@@ -1136,7 +1143,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
: new TPushPipeMetaResp()
.setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
.setExceptionMessages(Collections.singletonList(exceptionMessage));
- } catch (Exception e) {
+ } catch (final Exception e) {
LOGGER.error("Error occurred when pushing single pipe meta", e);
return new TPushPipeMetaResp()
.setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()));
@@ -1579,6 +1586,85 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
.getStatus());
}
+ @Override
+ public TSStatus constructTableDeviceBlackList(
+ final TTableDeviceDeletionWithPatternAndFilterReq req) {
+ final AtomicLong preDeletedNum = new AtomicLong(0);
+ final TSStatus executionResult =
+ executeSchemaBlackListTask(
+ req.getRegionIdList(),
+ consensusGroupId -> {
+ final TSStatus status =
+ new RegionWriteExecutor()
+ .execute(
+ new SchemaRegionId(consensusGroupId.getId()),
+ new ConstructTableDevicesBlackListNode(
+ new PlanNodeId(""),
+ req.getTableName(),
+ req.getPatternInfo(),
+ req.getFilterInfo()))
+ .getStatus();
+ if (status.code == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ preDeletedNum.getAndAdd(Long.parseLong(status.getMessage()));
+ }
+ return status;
+ });
+ executionResult.setMessage(String.valueOf(preDeletedNum.get()));
+ return executionResult;
+ }
+
+ @Override
+ public TSStatus rollbackTableDeviceBlackList(final TTableDeviceDeletionWithPatternReq req) {
+ return executeInternalSchemaTask(
+ req.getSchemaRegionIdList(),
+ consensusGroupId ->
+ new RegionWriteExecutor()
+ .execute(
+ new SchemaRegionId(consensusGroupId.getId()),
+ new RollbackTableDevicesBlackListNode(
+ new PlanNodeId(""), req.getTableName(), req.getPatternInfo()))
+ .getStatus());
+ }
+
+ @Override
+ public TSStatus invalidateMatchedTableDeviceCache(final TTableDeviceInvalidateCacheReq req) {
+ DataNodeSchemaLockManager.getInstance().takeWriteLock(SchemaLockType.VALIDATE_VS_DELETION);
+ try {
+ TableDeviceSchemaCache.getInstance()
+ .invalidate(
+ req.getDatabase(),
+ req.getTableName(),
+ DeleteDevice.constructPaths(
+ req.getDatabase(), req.getTableName(), req.getPatternInfo()));
+ return StatusUtils.OK;
+ } finally {
+ DataNodeSchemaLockManager.getInstance().releaseWriteLock(SchemaLockType.VALIDATE_VS_DELETION);
+ }
+ }
+
+ @Override
+ public TSStatus deleteDataForTableDevice(final TTableDeviceDeletionWithPatternAndFilterReq req) {
+ return executeInternalSchemaTask(
+ req.getRegionIdList(),
+ consensusGroupId -> {
+ // TODO
+ return StatusUtils.OK;
+ });
+ }
+
+ @Override
+ public TSStatus deleteTableDeviceInBlackList(final TTableDeviceDeletionWithPatternReq req) {
+ return executeInternalSchemaTask(
+ req.getSchemaRegionIdList(),
+ consensusGroupId ->
+ new RegionWriteExecutor()
+ .execute(
+ new SchemaRegionId(consensusGroupId.getId()),
+ new DeleteTableDevicesInBlackListNode(
+ new PlanNodeId(""), req.getTableName(), req.getPatternInfo()))
+ .getStatus());
+ }
+
public TTestConnectionResp submitTestConnectionTask(TNodeLocations nodeLocations) {
return new TTestConnectionResp(
new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()),
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceAttributeUpdater.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceAttributeUpdater.java
index 06f7919719d..db661300c14 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceAttributeUpdater.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceAttributeUpdater.java
@@ -25,7 +25,6 @@ import org.apache.iotdb.db.queryengine.transformation.dag.column.ColumnTransform
import org.apache.iotdb.db.queryengine.transformation.dag.column.leaf.LeafColumnTransformer;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.IMemMNode;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.TableDeviceInfo;
-import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.impl.ShowDevicesResult;
import org.apache.ratis.util.function.TriConsumer;
import org.apache.tsfile.block.column.Column;
@@ -44,11 +43,10 @@ import java.util.stream.Collectors;
import static org.apache.iotdb.db.queryengine.execution.operator.process.FilterAndProjectOperator.constructFilteredTsBlock;
-public class DeviceAttributeUpdater extends DevicePredicateHandler {
+public class DeviceAttributeUpdater extends DeviceUpdater {
private final List<ColumnTransformer> commonTransformerList;
private final List<LeafColumnTransformer> projectLeafColumnTransformerList;
private final List<ColumnTransformer> projectOutputTransformerList;
- final BiFunction<Integer, String, Binary> attributeProvider;
private final TriConsumer<String[], Integer, Object[]> attributeUpdater;
private final TsBlockBuilder filterTsBlockBuilder;
private final List<Integer> attributePointers = new ArrayList<>();
@@ -71,32 +69,22 @@ public class DeviceAttributeUpdater extends DevicePredicateHandler {
filterOutputTransformer,
database,
tableName,
- columnHeaderList);
+ columnHeaderList,
+ attributeProvider);
this.commonTransformerList = commonTransformerList;
this.projectLeafColumnTransformerList = projectLeafColumnTransformerList;
this.projectOutputTransformerList = projectOutputTransformerList;
- this.attributeProvider = attributeProvider;
this.attributeUpdater = attributeUpdater;
this.filterTsBlockBuilder = new TsBlockBuilder(filterOutputDataTypes);
}
public void handleDeviceNode(final IDeviceMNode<IMemMNode> node) {
- final ShowDevicesResult result =
- new ShowDevicesResult(
- null,
- node.isAlignedNullable(),
- node.getSchemaTemplateId(),
- node.getPartialPath().getNodes());
- final int pointer = ((TableDeviceInfo<IMemMNode>) node.getDeviceInfo()).getAttributePointer();
- result.setAttributeProvider(k -> attributeProvider.apply(pointer, k));
- attributePointers.add(pointer);
- addBatch(result);
- if (hasComputedResult()) {
- update();
- }
+ attributePointers.add(
+ ((TableDeviceInfo<IMemMNode>) node.getDeviceInfo()).getAttributePointer());
+ super.handleDeviceNode(node);
}
- private void update() {
+ protected void update() {
final TsBlock block = getFilterTsBlock();
projectLeafColumnTransformerList.forEach(
@@ -149,10 +137,6 @@ public class DeviceAttributeUpdater extends DevicePredicateHandler {
@Override
public void close() {
- prepareBatchResult();
- if (hasComputedResult()) {
- update();
- }
super.close();
projectOutputTransformerList.forEach(ColumnTransformer::close);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceBlackListConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceBlackListConstructor.java
new file mode 100644
index 00000000000..59d9cacc202
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceBlackListConstructor.java
@@ -0,0 +1,85 @@
+/*
+ * 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.execution.operator.schema.source;
+
+import org.apache.iotdb.commons.schema.node.role.IDeviceMNode;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.ColumnTransformer;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.leaf.LeafColumnTransformer;
+import org.apache.iotdb.db.schemaengine.rescon.MemSchemaRegionStatistics;
+import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.IMemMNode;
+
+import org.apache.tsfile.utils.Binary;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.BiFunction;
+
+public class DeviceBlackListConstructor extends DeviceUpdater {
+ private final List<IDeviceMNode<IMemMNode>> nodes =
+ new ArrayList<>(DEFAULT_MAX_TS_BLOCK_LINE_NUMBER);
+
+ private long preDeletedNum = 0;
+ private final MemSchemaRegionStatistics regionStatistics;
+
+ public DeviceBlackListConstructor(
+ final List<LeafColumnTransformer> filterLeafColumnTransformerList,
+ final ColumnTransformer filterOutputTransformer,
+ final String database,
+ final String tableName,
+ final List<ColumnHeader> columnHeaderList,
+ final BiFunction<Integer, String, Binary> attributeProvider,
+ final MemSchemaRegionStatistics regionStatistics) {
+ super(
+ filterLeafColumnTransformerList,
+ filterOutputTransformer,
+ database,
+ tableName,
+ columnHeaderList,
+ attributeProvider);
+ this.regionStatistics = regionStatistics;
+ }
+
+ public void handleDeviceNode(final IDeviceMNode<IMemMNode> node) {
+ if (withoutFilter()) {
+ node.preDeactivateSelfOrTemplate();
+ preDeletedNum++;
+ regionStatistics.decreaseTableDevice(tableName, 1);
+ return;
+ }
+ nodes.add(node);
+ super.handleDeviceNode(node);
+ }
+
+ @Override
+ protected void update() {
+ preDeletedNum += indexes.size();
+ regionStatistics.decreaseTableDevice(tableName, indexes.size());
+ for (final Integer index : indexes) {
+ nodes.get(index).preDeactivateSelfOrTemplate();
+ }
+ nodes.clear();
+ super.clear();
+ }
+
+ public long getPreDeletedNum() {
+ return preDeletedNum;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandler.java
index 5db32686ff9..b6781247717 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandler.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DevicePredicateHandler.java
@@ -43,14 +43,14 @@ public abstract class DevicePredicateHandler implements AutoCloseable {
protected final ColumnTransformer filterOutputTransformer;
private final List<TSDataType> inputDataTypes;
private final String database;
- private final String tableName;
+ protected final String tableName;
private final List<ColumnHeader> columnHeaderList;
// Batch logic
- private static final int DEFAULT_MAX_TS_BLOCK_SIZE_IN_BYTES =
- TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes();
+ protected static final int DEFAULT_MAX_TS_BLOCK_LINE_NUMBER =
+ TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber();
protected final List<IDeviceSchemaInfo> deviceSchemaBatch =
- new ArrayList<>(DEFAULT_MAX_TS_BLOCK_SIZE_IN_BYTES);
+ new ArrayList<>(DEFAULT_MAX_TS_BLOCK_LINE_NUMBER);
protected final List<Integer> indexes = new ArrayList<>();
protected TsBlock curBlock;
@@ -73,7 +73,7 @@ public abstract class DevicePredicateHandler implements AutoCloseable {
public void addBatch(final IDeviceSchemaInfo deviceSchemaInfo) {
deviceSchemaBatch.add(deviceSchemaInfo);
- if (deviceSchemaBatch.size() >= DEFAULT_MAX_TS_BLOCK_SIZE_IN_BYTES) {
+ if (deviceSchemaBatch.size() >= DEFAULT_MAX_TS_BLOCK_LINE_NUMBER) {
prepareBatchResult();
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceUpdater.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceUpdater.java
new file mode 100644
index 00000000000..5c085d843f2
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceUpdater.java
@@ -0,0 +1,80 @@
+/*
+ * 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.execution.operator.schema.source;
+
+import org.apache.iotdb.commons.schema.node.role.IDeviceMNode;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.ColumnTransformer;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.leaf.LeafColumnTransformer;
+import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.IMemMNode;
+import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.TableDeviceInfo;
+import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.impl.ShowDevicesResult;
+
+import org.apache.tsfile.utils.Binary;
+
+import java.util.List;
+import java.util.function.BiFunction;
+
+public abstract class DeviceUpdater extends DevicePredicateHandler {
+
+ private final BiFunction<Integer, String, Binary> attributeProvider;
+
+ protected DeviceUpdater(
+ final List<LeafColumnTransformer> filterLeafColumnTransformerList,
+ final ColumnTransformer filterOutputTransformer,
+ final String database,
+ final String tableName,
+ final List<ColumnHeader> columnHeaderList,
+ final BiFunction<Integer, String, Binary> attributeProvider) {
+ super(
+ filterLeafColumnTransformerList,
+ filterOutputTransformer,
+ database,
+ tableName,
+ columnHeaderList);
+ this.attributeProvider = attributeProvider;
+ }
+
+ public void handleDeviceNode(final IDeviceMNode<IMemMNode> node) {
+ final ShowDevicesResult result =
+ new ShowDevicesResult(
+ null,
+ node.isAlignedNullable(),
+ node.getSchemaTemplateId(),
+ node.getPartialPath().getNodes());
+ final int pointer = ((TableDeviceInfo<IMemMNode>) node.getDeviceInfo()).getAttributePointer();
+ result.setAttributeProvider(k -> attributeProvider.apply(pointer, k));
+ addBatch(result);
+ if (hasComputedResult()) {
+ update();
+ }
+ }
+
+ protected abstract void update();
+
+ @Override
+ public void close() {
+ prepareBatchResult();
+ if (hasComputedResult()) {
+ update();
+ }
+ super.close();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
index 24063d89dc3..88d8c59d170 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
@@ -51,6 +51,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AddColumn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ClearCache;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateTable;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DeleteDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescribeTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropTable;
@@ -315,13 +316,13 @@ public class Coordinator {
}
private IQueryExecution createQueryExecutionForTableModel(
- org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Statement statement,
- SqlParser sqlParser,
- IClientSession clientSession,
- MPPQueryContext queryContext,
- Metadata metadata,
- long timeOut,
- long startTime) {
+ final org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Statement statement,
+ final SqlParser sqlParser,
+ final IClientSession clientSession,
+ final MPPQueryContext queryContext,
+ final Metadata metadata,
+ final long timeOut,
+ final long startTime) {
queryContext.setTableQuery(true);
queryContext.setTimeOut(timeOut);
queryContext.setStartTime(startTime);
@@ -335,6 +336,7 @@ public class Coordinator {
|| statement instanceof AddColumn
|| statement instanceof SetProperties
|| statement instanceof DropTable
+ || statement instanceof DeleteDevice
|| statement instanceof ShowCluster
|| statement instanceof ShowRegions
|| statement instanceof ShowDataNodes
@@ -360,7 +362,7 @@ public class Coordinator {
if (statement instanceof WrappedInsertStatement) {
((WrappedInsertStatement) statement).setContext(queryContext);
}
- TableModelPlanner tableModelPlanner =
+ final TableModelPlanner tableModelPlanner =
new TableModelPlanner(
statement,
sqlParser,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCache.java
index 335439e62fc..f6e20ad58c7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCache.java
@@ -122,6 +122,9 @@ public interface IDualKeyCache<FK, SK, V> {
@GuardedBy("DataNodeSchemaCache#writeLock")
void invalidate(final FK firstKey, final SK secondKey);
+ @GuardedBy("DataNodeSchemaCache#writeLock")
+ void invalidate(final FK firstKey, final Predicate<SK> secondKeyChecker);
+
/** remove all entries matching the firstKey and the secondKey */
@GuardedBy("DataNodeSchemaCache#writeLock")
void invalidate(final Predicate<FK> firstKeyChecker, final Predicate<SK> secondKeyChecker);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java
index bbd44fddd44..126f5314495 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java
@@ -400,12 +400,46 @@ class DualKeyCacheImpl<FK, SK, V, T extends ICacheEntry<SK, V>>
return cacheEntryGroup;
});
+ cacheStats.decreaseMemoryUsage(usedMemorySize.get());
+ }
+
+ @Override
+ public void invalidate(final FK firstKey, final Predicate<SK> secondKeyChecker) {
+ final AtomicInteger estimateSize = new AtomicInteger(0);
+ firstKeyMap.compute(
+ firstKey,
+ (key, cacheEntryGroup) -> {
+ if (cacheEntryGroup == null) {
+ // has been removed by other threads
+ return null;
+ }
+
+ for (final Iterator<Map.Entry<SK, T>> it = cacheEntryGroup.getAllCacheEntries();
+ it.hasNext(); ) {
+ final Map.Entry<SK, T> entry = it.next();
+ if (cacheEntryManager.invalidate(entry.getValue())) {
+ cacheEntryGroup.removeCacheEntry(entry.getKey());
+ estimateSize.addAndGet(
+ sizeComputer.computeSecondKeySize(entry.getKey())
+ + sizeComputer.computeValueSize(entry.getValue().getValue()));
+ }
+ }
+
+ if (cacheEntryGroup.isEmpty()) {
+ estimateSize.getAndAdd(sizeComputer.computeFirstKeySize(firstKey));
+ return null;
+ }
+
+ return cacheEntryGroup;
+ });
+
+ cacheStats.decreaseMemoryUsage(estimateSize.get());
}
@Override
public void invalidate(
final Predicate<FK> firstKeyChecker, final Predicate<SK> secondKeyChecker) {
- int estimateSize = 0;
+ final AtomicInteger estimateSize = new AtomicInteger(0);
for (final FK firstKey : firstKeyMap.getAllKeys()) {
if (!firstKeyChecker.test(firstKey)) {
continue;
@@ -419,13 +453,22 @@ class DualKeyCacheImpl<FK, SK, V, T extends ICacheEntry<SK, V>>
if (cacheEntryManager.invalidate(entry.getValue())) {
entryGroup.removeCacheEntry(entry.getKey());
- estimateSize +=
+ estimateSize.addAndGet(
sizeComputer.computeSecondKeySize(entry.getKey())
- + sizeComputer.computeValueSize(entry.getValue().getValue());
+ + sizeComputer.computeValueSize(entry.getValue().getValue()));
}
}
+ firstKeyMap.compute(
+ firstKey,
+ (fk, sk) -> {
+ if (sk.isEmpty()) {
+ estimateSize.getAndAdd(sizeComputer.computeFirstKeySize(firstKey));
+ return null;
+ }
+ return sk;
+ });
}
- cacheStats.decreaseMemoryUsage(estimateSize);
+ cacheStats.decreaseMemoryUsage(estimateSize.get());
}
/**
@@ -474,6 +517,7 @@ class DualKeyCacheImpl<FK, SK, V, T extends ICacheEntry<SK, V>>
return map;
}
+ // Copied list, deletion-safe
List<K> getAllKeys() {
List<K> res = new ArrayList<>();
Arrays.stream(maps)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
index ea06a8bc793..5ba4e03b783 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.protocol.session.IClientSession;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.execution.warnings.WarningCollector;
import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.CreatePipePluginTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.DropPipePluginTask;
@@ -44,6 +45,7 @@ import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.ClearCacheTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.CreateDBTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.CreateTableTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.DeleteDeviceTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.DescribeTableTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.DropDBTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.DropTableTask;
@@ -67,8 +69,11 @@ import org.apache.iotdb.db.queryengine.plan.execution.config.sys.pipe.DropPipeTa
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.pipe.ShowPipeTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.pipe.StartPipeTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.pipe.StopPipeTask;
+import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analyzer;
+import org.apache.iotdb.db.queryengine.plan.relational.analyzer.StatementAnalyzerFactory;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableHeaderSchemaValidator;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.TableModelPlanner;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AddColumn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterPipe;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AstVisitor;
@@ -79,6 +84,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreatePipe;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreatePipePlugin;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DataType;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DeleteDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescribeTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropColumn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropDB;
@@ -480,6 +486,19 @@ public class TableConfigTaskVisitor extends AstVisitor<IConfigTask, MPPQueryCont
node.isExists());
}
+ @Override
+ protected IConfigTask visitDeleteDevice(final DeleteDevice node, final MPPQueryContext context) {
+ new Analyzer(
+ context,
+ context.getSession(),
+ new StatementAnalyzerFactory(metadata, null, new TableModelPlanner.NopAccessControl()),
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ WarningCollector.NOOP)
+ .analyze(node);
+ return new DeleteDeviceTask(node, context.getQueryId().getId(), context.getSession());
+ }
+
@Override
protected IConfigTask visitShowTables(final ShowTables node, final MPPQueryContext context) {
context.setQueryType(QueryType.READ);
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 8bdf44b4a4b..1ecd281a761 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
@@ -81,6 +81,8 @@ import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.confignode.rpc.thrift.TDeactivateSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabasesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceResp;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq;
@@ -144,6 +146,7 @@ import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
import org.apache.iotdb.db.protocol.client.DataNodeClientPoolFactory;
import org.apache.iotdb.db.protocol.session.IClientSession;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
@@ -172,6 +175,7 @@ import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowTTLTas
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowTriggersTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowVariablesTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.model.ShowModelsTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.DeleteDeviceTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.DescribeTableTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.ShowDBTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.ShowTablesDetailsTask;
@@ -193,6 +197,7 @@ import org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription.Sh
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
import org.apache.iotdb.db.queryengine.plan.expression.visitor.TransformToViewExpressionVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.view.AlterLogicalViewNode;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DeleteDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCluster;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDB;
@@ -3555,6 +3560,71 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
return future;
}
+ @Override
+ public SettableFuture<ConfigTaskResult> deleteDevice(
+ final DeleteDevice deleteDevice, final String queryId, final SessionInfo sessionInfo) {
+ final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try (final ConfigNodeClient client =
+ CLUSTER_DELETION_CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+
+ final ByteArrayOutputStream patternStream = new ByteArrayOutputStream();
+ try (final DataOutputStream outputStream = new DataOutputStream(patternStream)) {
+ deleteDevice.serializePatternInfo(outputStream);
+ } catch (final IOException ignored) {
+ // ByteArrayOutputStream won't throw IOException
+ }
+
+ final ByteArrayOutputStream filterStream = new ByteArrayOutputStream();
+ try (final DataOutputStream outputStream = new DataOutputStream(filterStream)) {
+ deleteDevice.serializeFilterInfo(outputStream, sessionInfo);
+ } catch (final IOException ignored) {
+ // ByteArrayOutputStream won't throw IOException
+ }
+
+ final TDeleteTableDeviceReq req =
+ new TDeleteTableDeviceReq(
+ PathUtils.qualifyDatabaseName(deleteDevice.getDatabase()),
+ deleteDevice.getTableName(),
+ queryId,
+ ByteBuffer.wrap(patternStream.toByteArray()),
+ ByteBuffer.wrap(filterStream.toByteArray()));
+
+ TDeleteTableDeviceResp resp;
+ do {
+ try {
+ resp = client.deleteDevice(req);
+ } catch (final TTransportException e) {
+ if (e.getType() == TTransportException.TIMED_OUT
+ || e.getCause() instanceof SocketTimeoutException) {
+ // time out mainly caused by slow execution, wait until
+ resp =
+ new TDeleteTableDeviceResp(
+ RpcUtils.getStatus(TSStatusCode.OVERLAP_WITH_EXISTING_TASK));
+ } else {
+ throw e;
+ }
+ }
+ // keep waiting until task ends
+ } while (TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode()
+ == resp.getStatus().getCode());
+
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() == resp.getStatus().getCode()) {
+ DeleteDeviceTask.buildTSBlock(resp.getDeletedNum(), future);
+ } else {
+ LOGGER.warn(
+ "Failed to delete devices from table {}.{}, status is {}.",
+ deleteDevice.getDatabase(),
+ deleteDevice.getTableName(),
+ resp.getStatus());
+ future.setException(
+ new IoTDBException(resp.getStatus().getMessage(), resp.getStatus().getCode()));
+ }
+ } catch (final ClientManagerException | TException e) {
+ future.setException(e);
+ }
+ return future;
+ }
+
private TSStatus sendAlterReq2ConfigNode(
final String database,
final String tableName,
@@ -3562,7 +3632,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
final AlterOrDropTableOperationType type,
final byte[] updateInfo,
final ConfigNodeClient client)
- throws ClientManagerException, TException {
+ throws TException {
TSStatus tsStatus;
final TAlterOrDropTableReq req = new TAlterOrDropTableReq();
req.setDatabase(database);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
index 98d57aa9ffc..b6baf99fb92 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
@@ -32,8 +32,10 @@ import org.apache.iotdb.confignode.rpc.thrift.TSpaceQuotaResp;
import org.apache.iotdb.confignode.rpc.thrift.TThrottleQuotaResp;
import org.apache.iotdb.db.protocol.session.IClientSession;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.view.AlterLogicalViewNode;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DeleteDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCluster;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDB;
@@ -336,6 +338,9 @@ public interface IConfigTaskExecutor {
SettableFuture<ConfigTaskResult> dropTable(
final String database, final String tableName, final String queryId, final boolean ifExists);
+ SettableFuture<ConfigTaskResult> deleteDevice(
+ final DeleteDevice deleteDevice, final String queryId, final SessionInfo sessionInfo);
+
SettableFuture<ConfigTaskResult> showVersion();
SettableFuture<ConfigTaskResult> showCurrentSqlDialect(String sqlDialect);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/AbstractAlterOrDropTableTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/AbstractAlterOrDropTableTask.java
index 9be925e0425..2ece21dee3a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/AbstractAlterOrDropTableTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/AbstractAlterOrDropTableTask.java
@@ -33,9 +33,11 @@ abstract class AbstractAlterOrDropTableTask implements IConfigTask {
protected final boolean tableIfExists;
protected AbstractAlterOrDropTableTask(
- String database, final String tableName, final String queryId, final boolean tableIfExists) {
- database = PathUtils.qualifyDatabaseName(database);
- this.database = database;
+ final String database,
+ final String tableName,
+ final String queryId,
+ final boolean tableIfExists) {
+ this.database = PathUtils.qualifyDatabaseName(database);
this.tableName = tableName;
this.queryId = queryId;
this.tableIfExists = tableIfExists;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/DeleteDeviceTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/DeleteDeviceTask.java
new file mode 100644
index 00000000000..01b3b31a156
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/DeleteDeviceTask.java
@@ -0,0 +1,72 @@
+/*
+ * 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.execution.config.metadata.relational;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.queryengine.common.SessionInfo;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
+import org.apache.iotdb.db.queryengine.common.header.DatasetHeader;
+import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DeleteDevice;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+
+import java.util.Collections;
+
+public class DeleteDeviceTask implements IConfigTask {
+ private final DeleteDevice deleteDevice;
+ private final String queryId;
+ private final SessionInfo sessionInfo;
+
+ public DeleteDeviceTask(
+ final DeleteDevice deleteDevice, final String queryId, final SessionInfo sessionInfo) {
+ this.deleteDevice = deleteDevice;
+ this.queryId = queryId;
+ this.sessionInfo = sessionInfo;
+ }
+
+ @Override
+ public ListenableFuture<ConfigTaskResult> execute(final IConfigTaskExecutor configTaskExecutor)
+ throws InterruptedException {
+ return configTaskExecutor.deleteDevice(deleteDevice, queryId, sessionInfo);
+ }
+
+ public static void buildTSBlock(
+ final long deviceNum, final SettableFuture<ConfigTaskResult> future) {
+ final TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT64));
+ builder.getTimeColumnBuilder().writeLong(0L);
+ builder.getColumnBuilder(0).writeLong(deviceNum);
+ builder.declarePosition();
+ future.set(
+ new ConfigTaskResult(
+ TSStatusCode.SUCCESS_STATUS,
+ builder.build(),
+ new DatasetHeader(
+ Collections.singletonList(
+ new ColumnHeader(IoTDBConstant.COLUMN_DELETED_DEVICE_NUM, TSDataType.INT64)),
+ true)));
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
index 8d67baf49a5..6c2ccb750f3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
@@ -420,7 +420,6 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
final Map<Symbol, List<InputLocation>> outputMappings = new LinkedHashMap<>();
int tsBlockIndex = 0;
for (final PlanNode childNode : children) {
-
int valueColumnIndex = 0;
for (final Symbol columnName : childNode.getOutputSymbols()) {
outputMappings
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
index 6c055d99f46..281c4749f69 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
@@ -119,8 +119,11 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LinearFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.PreviousFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceFetchNode;
@@ -248,6 +251,9 @@ public enum PlanNodeType {
TABLE_DEVICE_ATTRIBUTE_UPDATE((short) 907),
TABLE_DEVICE_ATTRIBUTE_COMMIT((short) 908),
TABLE_DEVICE_LOCATION_ADD((short) 909),
+ CONSTRUCT_TABLE_DEVICES_BLACK_LIST((short) 910),
+ ROLLBACK_TABLE_DEVICES_BLACK_LIST((short) 911),
+ DELETE_TABLE_DEVICES_IN_BLACK_LIST((short) 912),
TABLE_SCAN_NODE((short) 1000),
TABLE_FILTER_NODE((short) 1001),
@@ -553,6 +559,12 @@ public enum PlanNodeType {
return TableDeviceAttributeCommitUpdateNode.deserialize(buffer);
case 909:
return TableNodeLocationAddNode.deserialize(buffer);
+ case 910:
+ return ConstructTableDevicesBlackListNode.deserialize(buffer);
+ case 911:
+ return RollbackTableDevicesBlackListNode.deserialize(buffer);
+ case 912:
+ return DeleteTableDevicesInBlackListNode.deserialize(buffer);
case 1000:
return org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode
.deserialize(buffer);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
index 23f7749bfd1..55d4f374a78 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
@@ -125,8 +125,11 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LinearFillNo
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.PreviousFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceFetchNode;
@@ -544,6 +547,21 @@ public abstract class PlanVisitor<R, C> {
return visitPlan(node, context);
}
+ public R visitConstructTableDevicesBlackList(
+ final ConstructTableDevicesBlackListNode node, final C context) {
+ return visitPlan(node, context);
+ }
+
+ public R visitRollbackTableDevicesBlackList(
+ final RollbackTableDevicesBlackListNode node, final C context) {
+ return visitPlan(node, context);
+ }
+
+ public R visitDeleteTableDevicesInBlackList(
+ final DeleteTableDevicesInBlackListNode node, final C context) {
+ return visitPlan(node, context);
+ }
+
/////////////////////////////////////////////////////////////////////////////////////////////////
// Data Write Node
/////////////////////////////////////////////////////////////////////////////////////////////////
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
index a6bfc10d2bf..e7438a8c82c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
@@ -59,6 +59,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreatePipe;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreatePipePlugin;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DeleteDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DereferenceExpression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescribeTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropColumn;
@@ -299,7 +300,8 @@ public class StatementAnalyzer {
|| node instanceof FetchDevice
|| node instanceof ShowDevice
|| node instanceof CountDevice
- || node instanceof Update) {
+ || node instanceof Update
+ || node instanceof DeleteDevice) {
return returnScope;
}
checkState(
@@ -446,6 +448,26 @@ public class StatementAnalyzer {
return null;
}
+ @Override
+ protected Scope visitDeleteDevice(final DeleteDevice node, final Optional<Scope> context) {
+ // Actually write, but will return the result
+ queryContext.setQueryType(QueryType.READ);
+ analyzeTraverseDevice(node, context, node.getWhere().isPresent());
+ final TsTable table =
+ DataNodeTableCache.getInstance().getTable(node.getDatabase(), node.getTableName());
+ node.parseRawExpression(
+ null,
+ table,
+ table.getColumnList().stream()
+ .filter(
+ columnSchema ->
+ columnSchema.getColumnCategory().equals(TsTableColumnCategory.ATTRIBUTE))
+ .map(TsTableColumnSchema::getColumnName)
+ .collect(Collectors.toList()),
+ queryContext);
+ return null;
+ }
+
@Override
protected Scope visitDropFunction(DropFunction node, Optional<Scope> context) {
throw new SemanticException("Drop Function statement is not supported yet.");
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzerFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzerFactory.java
index 155e0d3338e..47b03729a51 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzerFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzerFactory.java
@@ -35,7 +35,7 @@ public class StatementAnalyzerFactory {
private final AccessControl accessControl;
public StatementAnalyzerFactory(
- Metadata metadata, SqlParser sqlParser, AccessControl accessControl) {
+ final Metadata metadata, final SqlParser sqlParser, final AccessControl accessControl) {
this.metadata = requireNonNull(metadata, "plannerContext is null");
this.sqlParser = sqlParser;
this.accessControl = requireNonNull(accessControl, "accessControl is null");
@@ -46,11 +46,11 @@ public class StatementAnalyzerFactory {
}
public StatementAnalyzer createStatementAnalyzer(
- Analysis analysis,
- MPPQueryContext context,
- SessionInfo session,
- WarningCollector warningCollector,
- CorrelationSupport correlationSupport) {
+ final Analysis analysis,
+ final MPPQueryContext context,
+ final SessionInfo session,
+ final WarningCollector warningCollector,
+ final CorrelationSupport correlationSupport) {
return new StatementAnalyzer(
this,
analysis,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
index 87636df4da7..ffb8748a627 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache;
import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.ExtendedPartialPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternUtil;
import org.apache.iotdb.commons.service.metric.MetricService;
@@ -48,6 +49,7 @@ import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -56,6 +58,9 @@ import java.util.OptionalLong;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.ToIntFunction;
+import java.util.stream.Collectors;
+
+import static org.apache.iotdb.commons.conf.IoTDBConstant.ONE_LEVEL_PATH_WILDCARD;
/**
* The {@link TableDeviceSchemaCache} caches some of the devices and their: attributes of tables /
@@ -541,6 +546,56 @@ public class TableDeviceSchemaCache {
}
}
+ // The fuzzy filters are not considered because:
+ // 1. We can actually invalidate more cache entries than we need.
+ // 2. Constructing the filterOperators may require some time and complication
+ // 3. The fuzzy filters may contain attributes, which may not exist or be stale
+ public void invalidate(
+ final String database, final String tableName, final List<PartialPath> patterns) {
+ readWriteLock.writeLock().lock();
+ try {
+ final TableId firstKey = new TableId(database, tableName);
+ if (patterns.isEmpty()) {
+ dualKeyCache.invalidate(firstKey);
+ } else {
+ final List<PartialPath> multiMatchList =
+ patterns.stream()
+ .filter(
+ idFilter -> {
+ if (!idFilter.hasWildcard()) {
+ final IDeviceID deviceId =
+ IDeviceID.Factory.DEFAULT_FACTORY.create(
+ Arrays.copyOfRange(
+ idFilter.getNodes(), 2, idFilter.getNodeLength()));
+ dualKeyCache.invalidate(firstKey, deviceId);
+ return false;
+ }
+ return true;
+ })
+ .collect(Collectors.toList());
+
+ dualKeyCache.invalidate(
+ firstKey,
+ deviceId -> {
+ final String[] segments = (String[]) deviceId.getSegments();
+ for (int i = 1; i < segments.length; ++i) {
+ for (final PartialPath path : multiMatchList) {
+ final int pathIndex = i + 2;
+ if (path.getNodes()[pathIndex].equals(segments[i])
+ || path.getNodes()[pathIndex].equals(ONE_LEVEL_PATH_WILDCARD)
+ && ((ExtendedPartialPath) path).match(pathIndex, segments[i])) {
+ return true;
+ }
+ }
+ }
+ return false;
+ });
+ }
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
public void invalidateLastCache() {
readWriteLock.writeLock().lock();
try {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableModelPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableModelPlanner.java
index 893a4c51482..1b891dead32 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableModelPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableModelPlanner.java
@@ -81,14 +81,15 @@ public class TableModelPlanner implements IPlanner {
asyncInternalServiceClientManager;
public TableModelPlanner(
- Statement statement,
- SqlParser sqlParser,
- Metadata metadata,
- ExecutorService executor,
- ExecutorService writeOperationExecutor,
- ScheduledExecutorService scheduledExecutor,
- IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> syncInternalServiceClientManager,
- IClientManager<TEndPoint, AsyncDataNodeInternalServiceClient>
+ final Statement statement,
+ final SqlParser sqlParser,
+ final Metadata metadata,
+ final ExecutorService executor,
+ final ExecutorService writeOperationExecutor,
+ final ScheduledExecutorService scheduledExecutor,
+ final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
+ syncInternalServiceClientManager,
+ final IClientManager<TEndPoint, AsyncDataNodeInternalServiceClient>
asyncInternalServiceClientManager) {
this.statement = statement;
this.sqlParser = sqlParser;
@@ -101,42 +102,39 @@ public class TableModelPlanner implements IPlanner {
}
@Override
- public IAnalysis analyze(MPPQueryContext context) {
- StatementAnalyzerFactory statementAnalyzerFactory =
- new StatementAnalyzerFactory(metadata, sqlParser, accessControl);
-
- Analyzer analyzer =
- new Analyzer(
+ public IAnalysis analyze(final MPPQueryContext context) {
+ return new Analyzer(
context,
context.getSession(),
- statementAnalyzerFactory,
+ new StatementAnalyzerFactory(metadata, sqlParser, accessControl),
Collections.emptyList(),
Collections.emptyMap(),
- warningCollector);
- return analyzer.analyze(statement);
+ warningCollector)
+ .analyze(statement);
}
@Override
- public LogicalQueryPlan doLogicalPlan(IAnalysis analysis, MPPQueryContext context) {
+ public LogicalQueryPlan doLogicalPlan(final IAnalysis analysis, final MPPQueryContext context) {
return new TableLogicalPlanner(
context, metadata, context.getSession(), symbolAllocator, warningCollector)
.plan((Analysis) analysis);
}
@Override
- public DistributedQueryPlan doDistributionPlan(IAnalysis analysis, LogicalQueryPlan logicalPlan) {
+ public DistributedQueryPlan doDistributionPlan(
+ final IAnalysis analysis, final LogicalQueryPlan logicalPlan) {
return new TableDistributedPlanner((Analysis) analysis, symbolAllocator, logicalPlan).plan();
}
@Override
public IScheduler doSchedule(
- IAnalysis analysis,
- DistributedQueryPlan distributedPlan,
- MPPQueryContext context,
- QueryStateMachine stateMachine) {
- IScheduler scheduler;
+ final IAnalysis analysis,
+ final DistributedQueryPlan distributedPlan,
+ final MPPQueryContext context,
+ final QueryStateMachine stateMachine) {
+ final IScheduler scheduler;
- boolean isPipeEnrichedTsFileLoad =
+ final boolean isPipeEnrichedTsFileLoad =
statement instanceof PipeEnriched
&& ((PipeEnriched) statement).getInnerStatement() instanceof LoadTsFile;
if (statement instanceof LoadTsFile || isPipeEnrichedTsFileLoad) {
@@ -216,5 +214,5 @@ public class TableModelPlanner implements IPlanner {
}
}
- private static class NopAccessControl implements AccessControl {}
+ public static class NopAccessControl implements AccessControl {}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/AbstractTableDevicesDeletionNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/AbstractTableDevicesDeletionNode.java
new file mode 100644
index 00000000000..95e7cf2c9e1
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/AbstractTableDevicesDeletionNode.java
@@ -0,0 +1,91 @@
+/*
+ * 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.planner.node.schema;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegionPlan;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import javax.annotation.Nonnull;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+
+abstract class AbstractTableDevicesDeletionNode extends PlanNode implements ISchemaRegionPlan {
+
+ protected final String tableName;
+ protected final byte[] patternInfo;
+
+ protected AbstractTableDevicesDeletionNode(
+ final PlanNodeId id, final String tableName, final @Nonnull byte[] patternInfo) {
+ super(id);
+ this.tableName = tableName;
+ this.patternInfo = patternInfo;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public byte[] getPatternInfo() {
+ return patternInfo;
+ }
+
+ @Override
+ public List<PlanNode> getChildren() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void addChild(final PlanNode child) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int allowedChildCount() {
+ return 0;
+ }
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ return null;
+ }
+
+ @Override
+ protected void serializeAttributes(final ByteBuffer byteBuffer) {
+ getType().serialize(byteBuffer);
+ ReadWriteIOUtils.write(tableName, byteBuffer);
+ ReadWriteIOUtils.write(patternInfo.length, byteBuffer);
+ byteBuffer.put(patternInfo);
+ }
+
+ @Override
+ protected void serializeAttributes(final DataOutputStream stream) throws IOException {
+ getType().serialize(stream);
+ ReadWriteIOUtils.write(tableName, stream);
+ ReadWriteIOUtils.write(patternInfo.length, stream);
+ stream.write(patternInfo);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/ConstructTableDevicesBlackListNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/ConstructTableDevicesBlackListNode.java
new file mode 100644
index 00000000000..761bc2bb1fd
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/ConstructTableDevicesBlackListNode.java
@@ -0,0 +1,104 @@
+/*
+ * 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.planner.node.schema;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionPlanType;
+import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionPlanVisitor;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import javax.annotation.Nonnull;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class ConstructTableDevicesBlackListNode extends AbstractTableDevicesDeletionNode {
+ public static final ConstructTableDevicesBlackListNode MOCK_INSTANCE =
+ new ConstructTableDevicesBlackListNode(new PlanNodeId(""), null, new byte[0], new byte[0]);
+
+ private final byte[] filterInfo;
+
+ public ConstructTableDevicesBlackListNode(
+ final PlanNodeId id,
+ final String tableName,
+ final @Nonnull byte[] patternInfo,
+ final @Nonnull byte[] filterInfo) {
+ super(id, tableName, patternInfo);
+ this.filterInfo = filterInfo;
+ }
+
+ public byte[] getFilterInfo() {
+ return filterInfo;
+ }
+
+ @Override
+ public PlanNode clone() {
+ return new ConstructTableDevicesBlackListNode(id, tableName, patternInfo, filterInfo);
+ }
+
+ @Override
+ public PlanNodeType getType() {
+ return PlanNodeType.CONSTRUCT_TABLE_DEVICES_BLACK_LIST;
+ }
+
+ public static ConstructTableDevicesBlackListNode deserialize(final ByteBuffer buffer) {
+ final String tableName = ReadWriteIOUtils.readString(buffer);
+ final byte[] patternInfo = new byte[ReadWriteIOUtils.readInt(buffer)];
+ buffer.get(patternInfo);
+ final byte[] filterInfo = new byte[ReadWriteIOUtils.readInt(buffer)];
+ buffer.get(filterInfo);
+ final PlanNodeId planNodeId = PlanNodeId.deserialize(buffer);
+ return new ConstructTableDevicesBlackListNode(planNodeId, tableName, patternInfo, filterInfo);
+ }
+
+ @Override
+ protected void serializeAttributes(final ByteBuffer byteBuffer) {
+ super.serializeAttributes(byteBuffer);
+ ReadWriteIOUtils.write(filterInfo.length, byteBuffer);
+ byteBuffer.put(filterInfo);
+ }
+
+ @Override
+ protected void serializeAttributes(final DataOutputStream stream) throws IOException {
+ super.serializeAttributes(stream);
+ ReadWriteIOUtils.write(filterInfo.length, stream);
+ stream.write(filterInfo);
+ }
+
+ @Override
+ public <R, C> R accept(final PlanVisitor<R, C> visitor, final C context) {
+ return visitor.visitConstructTableDevicesBlackList(this, context);
+ }
+
+ @Override
+ public SchemaRegionPlanType getPlanType() {
+ return SchemaRegionPlanType.CONSTRUCT_TABLE_DEVICES_BLACK_LIST;
+ }
+
+ @Override
+ public <R, C> R accept(final SchemaRegionPlanVisitor<R, C> visitor, final C context) {
+ return visitor.visitConstructTableDevicesBlackList(this, context);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/CreateOrUpdateTableDeviceNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/CreateOrUpdateTableDeviceNode.java
index bba380559cd..efc5c8f9278 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/CreateOrUpdateTableDeviceNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/CreateOrUpdateTableDeviceNode.java
@@ -290,7 +290,7 @@ public class CreateOrUpdateTableDeviceNode extends WritePlanNode implements ISch
}
@Override
- public boolean equals(Object o) {
+ public boolean equals(final Object o) {
if (this == o) {
return true;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/DeleteTableDevicesInBlackListNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/DeleteTableDevicesInBlackListNode.java
new file mode 100644
index 00000000000..5b51af4af16
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/DeleteTableDevicesInBlackListNode.java
@@ -0,0 +1,76 @@
+/*
+ * 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.planner.node.schema;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionPlanType;
+import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionPlanVisitor;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import javax.annotation.Nonnull;
+
+import java.nio.ByteBuffer;
+
+public class DeleteTableDevicesInBlackListNode extends AbstractTableDevicesDeletionNode {
+ public static final DeleteTableDevicesInBlackListNode MOCK_INSTANCE =
+ new DeleteTableDevicesInBlackListNode(new PlanNodeId(""), null, new byte[0]);
+
+ public DeleteTableDevicesInBlackListNode(
+ final PlanNodeId id, final String tableName, final @Nonnull byte[] patternInfo) {
+ super(id, tableName, patternInfo);
+ }
+
+ @Override
+ public PlanNode clone() {
+ return new DeleteTableDevicesInBlackListNode(id, tableName, patternInfo);
+ }
+
+ @Override
+ public PlanNodeType getType() {
+ return PlanNodeType.DELETE_TABLE_DEVICES_IN_BLACK_LIST;
+ }
+
+ public static DeleteTableDevicesInBlackListNode deserialize(final ByteBuffer buffer) {
+ final String tableName = ReadWriteIOUtils.readString(buffer);
+ final byte[] updateBytes = new byte[ReadWriteIOUtils.readInt(buffer)];
+ buffer.get(updateBytes);
+ final PlanNodeId planNodeId = PlanNodeId.deserialize(buffer);
+ return new DeleteTableDevicesInBlackListNode(planNodeId, tableName, updateBytes);
+ }
+
+ @Override
+ public <R, C> R accept(final PlanVisitor<R, C> visitor, final C context) {
+ return visitor.visitDeleteTableDevicesInBlackList(this, context);
+ }
+
+ @Override
+ public SchemaRegionPlanType getPlanType() {
+ return SchemaRegionPlanType.DELETE_TABLE_DEVICES_IN_BLACK_LIST;
+ }
+
+ @Override
+ public <R, C> R accept(final SchemaRegionPlanVisitor<R, C> visitor, final C context) {
+ return visitor.visitDeleteTableDevicesInBlackList(this, context);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/RollbackTableDevicesBlackListNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/RollbackTableDevicesBlackListNode.java
new file mode 100644
index 00000000000..b643e9b81e0
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/RollbackTableDevicesBlackListNode.java
@@ -0,0 +1,76 @@
+/*
+ * 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.planner.node.schema;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionPlanType;
+import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionPlanVisitor;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import javax.annotation.Nonnull;
+
+import java.nio.ByteBuffer;
+
+public class RollbackTableDevicesBlackListNode extends AbstractTableDevicesDeletionNode {
+ public static final RollbackTableDevicesBlackListNode MOCK_INSTANCE =
+ new RollbackTableDevicesBlackListNode(new PlanNodeId(""), null, new byte[0]);
+
+ public RollbackTableDevicesBlackListNode(
+ final PlanNodeId id, final String tableName, final @Nonnull byte[] updateBytes) {
+ super(id, tableName, updateBytes);
+ }
+
+ @Override
+ public PlanNode clone() {
+ return new RollbackTableDevicesBlackListNode(id, tableName, patternInfo);
+ }
+
+ @Override
+ public PlanNodeType getType() {
+ return PlanNodeType.ROLLBACK_TABLE_DEVICES_BLACK_LIST;
+ }
+
+ public static RollbackTableDevicesBlackListNode deserialize(final ByteBuffer buffer) {
+ final String tableName = ReadWriteIOUtils.readString(buffer);
+ final byte[] updateBytes = new byte[ReadWriteIOUtils.readInt(buffer)];
+ buffer.get(updateBytes);
+ final PlanNodeId planNodeId = PlanNodeId.deserialize(buffer);
+ return new RollbackTableDevicesBlackListNode(planNodeId, tableName, updateBytes);
+ }
+
+ @Override
+ public <R, C> R accept(final PlanVisitor<R, C> visitor, final C context) {
+ return visitor.visitRollbackTableDevicesBlackList(this, context);
+ }
+
+ @Override
+ public SchemaRegionPlanType getPlanType() {
+ return SchemaRegionPlanType.ROLLBACK_TABLE_DEVICES_BLACK_LIST;
+ }
+
+ @Override
+ public <R, C> R accept(final SchemaRegionPlanVisitor<R, C> visitor, final C context) {
+ return visitor.visitRollbackTableDevicesBlackList(this, context);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractTraverseDevice.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractTraverseDevice.java
index 2ed022815e5..2e50826f8e2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractTraverseDevice.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractTraverseDevice.java
@@ -60,10 +60,10 @@ public abstract class AbstractTraverseDevice extends Statement {
* <p>Each inner list represents a device pattern and each expression of it represents one
* condition on some id column.
*/
- private List<List<SchemaFilter>> idDeterminedFilterList;
+ protected List<List<SchemaFilter>> idDeterminedFilterList;
/** filters/conditions involving non-id columns and concat by OR to id column filters */
- private Expression idFuzzyPredicate;
+ protected Expression idFuzzyPredicate;
private List<IDeviceID> partitionKeyList;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java
index ffba78a3d40..fb8ebd09fac 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java
@@ -307,27 +307,31 @@ public abstract class AstVisitor<R, C> {
return visitNode(node, context);
}
- protected R visitCreateDB(CreateDB node, C context) {
+ protected R visitCreateDB(final CreateDB node, final C context) {
return visitStatement(node, context);
}
- protected R visitDropDB(DropDB node, C context) {
+ protected R visitDropDB(final DropDB node, final C context) {
return visitStatement(node, context);
}
- protected R visitShowDB(ShowDB node, C context) {
+ protected R visitShowDB(final ShowDB node, final C context) {
return visitStatement(node, context);
}
- protected R visitCreateTable(CreateTable node, C context) {
+ protected R visitCreateTable(final CreateTable node, final C context) {
return visitStatement(node, context);
}
- protected R visitProperty(Property node, C context) {
+ protected R visitProperty(final Property node, final C context) {
return visitNode(node, context);
}
- protected R visitDropTable(DropTable node, C context) {
+ protected R visitDropTable(final DropTable node, final C context) {
+ return visitStatement(node, context);
+ }
+
+ protected R visitDeleteDevice(final DeleteDevice node, final C context) {
return visitStatement(node, context);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/CreateTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/CreateTable.java
index caca3ff5795..c5942688e1f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/CreateTable.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/CreateTable.java
@@ -39,20 +39,6 @@ public class CreateTable extends Statement {
@Nullable private final String charsetName;
private final List<Property> properties;
- public CreateTable(
- final QualifiedName name,
- final List<ColumnDefinition> elements,
- final boolean ifNotExists,
- final @Nullable String charsetName,
- final List<Property> properties) {
- super(null);
- this.name = requireNonNull(name, "name is null");
- this.elements = ImmutableList.copyOf(requireNonNull(elements, "elements is null"));
- this.ifNotExists = ifNotExists;
- this.charsetName = charsetName;
- this.properties = requireNonNull(properties, "properties is null");
- }
-
public CreateTable(
final NodeLocation location,
final QualifiedName name,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DeleteDevice.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DeleteDevice.java
new file mode 100644
index 00000000000..9fd8471268c
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DeleteDevice.java
@@ -0,0 +1,198 @@
+/*
+ * 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.sql.ast;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.schema.filter.SchemaFilter;
+import org.apache.iotdb.db.queryengine.common.SessionInfo;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
+import org.apache.iotdb.db.queryengine.execution.operator.schema.source.DeviceBlackListConstructor;
+import org.apache.iotdb.db.queryengine.execution.operator.schema.source.TableDeviceQuerySource;
+import org.apache.iotdb.db.queryengine.execution.relational.ColumnTransformerBuilder;
+import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
+import org.apache.iotdb.db.queryengine.plan.planner.LocalExecutionPlanner;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.ColumnTransformer;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.leaf.LeafColumnTransformer;
+import org.apache.iotdb.db.schemaengine.rescon.MemSchemaRegionStatistics;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.tsfile.read.common.type.TypeFactory;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiFunction;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.MoreObjects.toStringHelper;
+
+public class DeleteDevice extends AbstractTraverseDevice {
+
+ public DeleteDevice(final NodeLocation location, final Table table, final Expression where) {
+ super(location, table, where);
+ }
+
+ public void serializePatternInfo(final DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(getIdDeterminedFilterList().size(), stream);
+ for (final List<SchemaFilter> filterList : idDeterminedFilterList) {
+ ReadWriteIOUtils.write(filterList.size(), stream);
+ for (final SchemaFilter filter : filterList) {
+ SchemaFilter.serialize(filter, stream);
+ }
+ }
+ }
+
+ public void serializeFilterInfo(final DataOutputStream stream, final SessionInfo sessionInfo)
+ throws IOException {
+ ReadWriteIOUtils.write(idFuzzyPredicate == null ? (byte) 0 : (byte) 1, stream);
+ if (idFuzzyPredicate != null) {
+ Expression.serialize(idFuzzyPredicate, stream);
+ }
+
+ ReadWriteIOUtils.write(columnHeaderList.size(), stream);
+ for (final ColumnHeader columnHeader : columnHeaderList) {
+ columnHeader.serialize(stream);
+ }
+
+ ReadWriteIOUtils.write(Objects.nonNull(sessionInfo), stream);
+ if (Objects.nonNull(sessionInfo)) {
+ sessionInfo.serialize(stream);
+ }
+ }
+
+ public static List<PartialPath> constructPaths(
+ final String database, final String tableName, final byte[] patternInfo) {
+ final ByteBuffer buffer = ByteBuffer.wrap(patternInfo);
+
+ // Device pattern list
+ int size = ReadWriteIOUtils.readInt(buffer);
+ final List<List<SchemaFilter>> idDeterminedFilterList = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ final int singleSize = ReadWriteIOUtils.readInt(buffer);
+ idDeterminedFilterList.add(new ArrayList<>(singleSize));
+ for (int k = 0; k < singleSize; k++) {
+ idDeterminedFilterList.get(i).add(SchemaFilter.deserialize(buffer));
+ }
+ }
+
+ return TableDeviceQuerySource.getDevicePatternList(database, tableName, idDeterminedFilterList);
+ }
+
+ public static DeviceBlackListConstructor constructDevicePredicateUpdater(
+ final String database,
+ final String tableName,
+ final byte[] filterInfo,
+ final BiFunction<Integer, String, Binary> attributeProvider,
+ final MemSchemaRegionStatistics regionStatistics) {
+ final ByteBuffer buffer = ByteBuffer.wrap(filterInfo);
+
+ Expression predicate = null;
+ if (buffer.get() == 1) {
+ predicate = Expression.deserialize(buffer);
+ }
+
+ final int size = ReadWriteIOUtils.readInt(buffer);
+ final List<ColumnHeader> columnHeaderList = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ columnHeaderList.add(ColumnHeader.deserialize(buffer));
+ }
+
+ SessionInfo sessionInfo = null;
+ if (ReadWriteIOUtils.readBool(buffer)) {
+ sessionInfo = SessionInfo.deserializeFrom(buffer);
+ }
+
+ final AtomicInteger valueColumnIndex = new AtomicInteger(0);
+ final Map<Symbol, List<InputLocation>> inputLocations =
+ columnHeaderList.stream()
+ .collect(
+ Collectors.toMap(
+ columnHeader -> new Symbol(columnHeader.getColumnName()),
+ columnHeader ->
+ Collections.singletonList(
+ new InputLocation(0, valueColumnIndex.getAndIncrement()))));
+
+ final TypeProvider mockTypeProvider =
+ new TypeProvider(
+ columnHeaderList.stream()
+ .collect(
+ Collectors.toMap(
+ columnHeader -> new Symbol(columnHeader.getColumnName()),
+ columnHeader -> TypeFactory.getType(columnHeader.getColumnType()))));
+ final Metadata metadata = LocalExecutionPlanner.getInstance().metadata;
+
+ // records LeafColumnTransformer of filter
+ final List<LeafColumnTransformer> filterLeafColumnTransformerList = new ArrayList<>();
+
+ // records subexpression -> ColumnTransformer for filter
+ final Map<Expression, ColumnTransformer> filterExpressionColumnTransformerMap = new HashMap<>();
+
+ final ColumnTransformerBuilder visitor = new ColumnTransformerBuilder();
+
+ final ColumnTransformer filterOutputTransformer =
+ Objects.nonNull(predicate)
+ ? visitor.process(
+ predicate,
+ new ColumnTransformerBuilder.Context(
+ sessionInfo,
+ filterLeafColumnTransformerList,
+ inputLocations,
+ filterExpressionColumnTransformerMap,
+ ImmutableMap.of(),
+ ImmutableList.of(),
+ ImmutableList.of(),
+ 0,
+ mockTypeProvider,
+ metadata))
+ : null;
+
+ return new DeviceBlackListConstructor(
+ filterLeafColumnTransformerList,
+ filterOutputTransformer,
+ database,
+ tableName,
+ columnHeaderList,
+ attributeProvider,
+ regionStatistics);
+ }
+
+ @Override
+ public <R, C> R accept(final AstVisitor<R, C> visitor, final C context) {
+ return visitor.visitDeleteDevice(this, context);
+ }
+
+ @Override
+ public String toString() {
+ return toStringHelper(this) + " - " + super.toStringContent();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
index 85d5638a0b0..3dc379b9e52 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
@@ -55,6 +55,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CurrentUser;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DataType;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DataTypeParameter;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DeleteDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DereferenceExpression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescribeTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DoubleLiteral;
@@ -569,6 +570,17 @@ public class AstBuilder extends RelationalSqlBaseVisitor<Node> {
: null);
}
+ @Override
+ public Node visitDeleteDeviceStatement(
+ final RelationalSqlParser.DeleteDeviceStatementContext ctx) {
+ return new DeleteDevice(
+ getLocation(ctx),
+ new Table(getLocation(ctx), getQualifiedName(ctx.qualifiedName())),
+ Objects.nonNull(ctx.booleanExpression())
+ ? (Expression) visit(ctx.booleanExpression())
+ : null);
+ }
+
@Override
public Node visitUpdateAssignment(RelationalSqlParser.UpdateAssignmentContext ctx) {
return new UpdateAssignment(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/rescon/MemSchemaRegionStatistics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/rescon/MemSchemaRegionStatistics.java
index 686795bde3b..042b06cadca 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/rescon/MemSchemaRegionStatistics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/rescon/MemSchemaRegionStatistics.java
@@ -107,6 +107,10 @@ public class MemSchemaRegionStatistics implements ISchemaRegionStatistics {
tableDeviceNumber.compute(table, (tableName, num) -> Objects.nonNull(num) ? num + 1 : 1L);
}
+ public void decreaseTableDevice(final String table, final long decrease) {
+ tableDeviceNumber.computeIfPresent(table, (tableName, num) -> num - decrease);
+ }
+
public void resetTableDevice(final String table) {
tableDeviceNumber.computeIfPresent(table, (tableName, num) -> 0L);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java
index 5787547ae56..a120443a086 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java
@@ -28,8 +28,11 @@ import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.exception.metadata.SchemaQuotaExceededException;
import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode;
@@ -338,6 +341,18 @@ public interface ISchemaRegion {
void deleteTableDevice(final DeleteTableDeviceNode deleteTableDeviceNode)
throws MetadataException;
+ long constructTableDevicesBlackList(
+ final ConstructTableDevicesBlackListNode constructDevicesBlackListNode)
+ throws MetadataException;
+
+ void rollbackTableDevicesBlackList(
+ final RollbackTableDevicesBlackListNode rollbackTableDevicesBlackListNode)
+ throws MetadataException;
+
+ void deleteTableDevicesInBlackList(
+ final DeleteTableDevicesInBlackListNode rollbackTableDevicesBlackListNode)
+ throws MetadataException;
+
// endregion
// region Interfaces for SchemaReader
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegionPlan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegionPlan.java
index f48d3902d0a..2f67487bfbc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegionPlan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegionPlan.java
@@ -23,5 +23,5 @@ public interface ISchemaRegionPlan {
SchemaRegionPlanType getPlanType();
- <R, C> R accept(SchemaRegionPlanVisitor<R, C> visitor, C context);
+ <R, C> R accept(final SchemaRegionPlanVisitor<R, C> visitor, final C context);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionPlanType.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionPlanType.java
index 0be34b48dfb..40e7e216a00 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionPlanType.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionPlanType.java
@@ -57,6 +57,9 @@ public enum SchemaRegionPlanType {
COMMIT_UPDATE_TABLE_DEVICE_ATTRIBUTE((byte) 102),
ADD_NODE_LOCATION((byte) 103),
DELETE_TABLE_DEVICE((byte) 104),
+ CONSTRUCT_TABLE_DEVICES_BLACK_LIST((byte) 105),
+ ROLLBACK_TABLE_DEVICES_BLACK_LIST((byte) 106),
+ DELETE_TABLE_DEVICES_IN_BLACK_LIST((byte) 107),
// query plan doesn't need any ser/deSer, thus use one type to represent all
READ_SCHEMA(Byte.MAX_VALUE);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionPlanVisitor.java
index d34d54cd962..5c432f792bb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionPlanVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionPlanVisitor.java
@@ -19,8 +19,11 @@
package org.apache.iotdb.db.schemaengine.schemaregion;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode;
@@ -154,4 +157,20 @@ public abstract class SchemaRegionPlanVisitor<R, C> {
final DeleteTableDeviceNode deleteTableDevicePlan, final C context) {
return visitSchemaRegionPlan(deleteTableDevicePlan, context);
}
+
+ public R visitConstructTableDevicesBlackList(
+ final ConstructTableDevicesBlackListNode constructTableDevicesBlackListPlan,
+ final C context) {
+ return visitSchemaRegionPlan(constructTableDevicesBlackListPlan, context);
+ }
+
+ public R visitRollbackTableDevicesBlackList(
+ final RollbackTableDevicesBlackListNode rollbackTableDevicesBlackListPlan, final C context) {
+ return visitSchemaRegionPlan(rollbackTableDevicesBlackListPlan, context);
+ }
+
+ public R visitDeleteTableDevicesInBlackList(
+ final DeleteTableDevicesInBlackListNode deleteTableDevicesInBlackListPlan, final C context) {
+ return visitSchemaRegionPlan(deleteTableDevicesInBlackListPlan, context);
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/DeviceAttributeCacheUpdater.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/DeviceAttributeCacheUpdater.java
index 45a7c022969..28f92bad165 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/DeviceAttributeCacheUpdater.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/DeviceAttributeCacheUpdater.java
@@ -124,6 +124,35 @@ public class DeviceAttributeCacheUpdater {
});
}
+ public void invalidate(final String tableName) {
+ attributeUpdateMap.forEach(
+ (location, container) -> {
+ final long size = container.invalidate(tableName);
+ releaseMemory(size);
+ updateContainerStatistics.computeIfPresent(
+ location,
+ (k, v) -> {
+ v.decreaseEntrySize(size);
+ return v;
+ });
+ });
+ }
+
+ // root.database.table.[deviceNodes]
+ public void invalidate(final String[] pathNodes) {
+ attributeUpdateMap.forEach(
+ (location, container) -> {
+ final long size = container.invalidate(pathNodes);
+ releaseMemory(size);
+ updateContainerStatistics.computeIfPresent(
+ location,
+ (k, v) -> {
+ v.decreaseEntrySize(size);
+ return v;
+ });
+ });
+ }
+
public Pair<Long, Map<TDataNodeLocation, byte[]>> getAttributeUpdateInfo(
final @Nonnull AtomicInteger limit, final @Nonnull AtomicBoolean hasRemaining) {
// Note that the "updateContainerStatistics" is unsafe to use here for whole read of detail
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateClearContainer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateClearContainer.java
index af5b76c449c..0de2d635a43 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateClearContainer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateClearContainer.java
@@ -78,6 +78,21 @@ public class UpdateClearContainer implements UpdateContainer {
return outputStream.toByteArray();
}
+ @Override
+ public long invalidate(final String tableName) {
+ if (tableNames.contains(tableName)) {
+ tableNames.remove(tableName);
+ return RamUsageEstimator.sizeOf(tableName);
+ }
+ return 0;
+ }
+
+ @Override
+ public long invalidate(final String[] pathNodes) {
+ // Do nothing
+ return 0;
+ }
+
private void serializeWithLimit(
final RewritableByteArrayOutputStream outputStream,
final AtomicInteger limitBytes,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateContainer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateContainer.java
index 0574676aa1f..6c1ae66bd0d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateContainer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateContainer.java
@@ -46,6 +46,10 @@ public interface UpdateContainer {
byte[] getUpdateContent(
final @Nonnull AtomicInteger limitBytes, final @Nonnull AtomicBoolean hasRemaining);
+ long invalidate(final String tableName);
+
+ long invalidate(final String[] pathNodes);
+
Pair<Long, Boolean> updateSelfByCommitContainer(final UpdateContainer commitContainer);
void serialize(final OutputStream outputstream) throws IOException;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateDetailContainer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateDetailContainer.java
index 1245cb3e3dd..ab4e238b2b2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateDetailContainer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateDetailContainer.java
@@ -44,6 +44,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
+// Note that all the writings to the containers are guarded by the schema region lock.
+// The readings are guarded byt the concurrent segment lock, thus there is no need
+// to worry about the concurrent safety.
@ThreadSafe
public class UpdateDetailContainer implements UpdateContainer {
@@ -124,6 +127,46 @@ public class UpdateDetailContainer implements UpdateContainer {
return outputStream.toByteArray();
}
+ @Override
+ public long invalidate(final String tableName) {
+ final AtomicLong result = new AtomicLong(0);
+ handleTableRemoval(tableName, result);
+ return result.get();
+ }
+
+ // pathNodes.length >= 3
+ @Override
+ public long invalidate(final @Nonnull String[] pathNodes) {
+ final AtomicLong result = new AtomicLong(0);
+ updateMap.compute(
+ pathNodes[2],
+ (name, value) -> {
+ if (Objects.isNull(value)) {
+ return null;
+ }
+ value.compute(
+ Arrays.asList(pathNodes).subList(3, pathNodes.length),
+ (device, attributes) -> {
+ if (Objects.nonNull(attributes)) {
+ result.addAndGet(
+ RamUsageEstimator.HASHTABLE_RAM_BYTES_PER_ENTRY
+ + sizeOfList(device)
+ + sizeOfMapEntries(attributes));
+ }
+ return null;
+ });
+ if (value.isEmpty()) {
+ result.addAndGet(
+ RamUsageEstimator.HASHTABLE_RAM_BYTES_PER_ENTRY
+ + RamUsageEstimator.sizeOf(name)
+ + MAP_SIZE);
+ return null;
+ }
+ return value;
+ });
+ return 0;
+ }
+
private void serializeWithLimit(
final RewritableByteArrayOutputStream outputStream,
final AtomicInteger limitBytes,
@@ -253,26 +296,24 @@ public class UpdateDetailContainer implements UpdateContainer {
} else {
((UpdateClearContainer) commitContainer)
.getTableNames()
- .forEach(
- tableName -> {
- final ConcurrentMap<List<String>, ConcurrentMap<String, Binary>> deviceMap =
- updateMap.remove(tableName);
- if (Objects.nonNull(deviceMap)) {
- result.addAndGet(
- deviceMap.size()
- * (RamUsageEstimator.HASHTABLE_RAM_BYTES_PER_ENTRY + MAP_SIZE)
- + deviceMap.entrySet().stream()
- .mapToLong(
- entry ->
- sizeOfList(entry.getKey())
- + sizeOfMapEntries(entry.getValue()))
- .reduce(0, Long::sum));
- }
- });
+ .forEach(tableName -> handleTableRemoval(tableName, result));
}
return new Pair<>(result.get(), updateMap.isEmpty());
}
+ private void handleTableRemoval(final String tableName, final AtomicLong result) {
+ final ConcurrentMap<List<String>, ConcurrentMap<String, Binary>> deviceMap =
+ updateMap.remove(tableName);
+ if (Objects.nonNull(deviceMap)) {
+ result.addAndGet(
+ deviceMap.size() * (RamUsageEstimator.HASHTABLE_RAM_BYTES_PER_ENTRY + MAP_SIZE)
+ + deviceMap.entrySet().stream()
+ .mapToLong(
+ entry -> sizeOfList(entry.getKey()) + sizeOfMapEntries(entry.getValue()))
+ .reduce(0, Long::sum));
+ }
+ }
+
private static long sizeOfList(final @Nonnull List<String> input) {
return input.stream().map(RamUsageEstimator::sizeOf).reduce(LIST_SIZE, Long::sum);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
index 7d3dd5fc326..829b5e4d960 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
@@ -44,6 +44,7 @@ import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
import org.apache.iotdb.db.queryengine.execution.operator.schema.source.DeviceAttributeUpdater;
+import org.apache.iotdb.db.queryengine.execution.operator.schema.source.DeviceBlackListConstructor;
import org.apache.iotdb.db.queryengine.execution.operator.schema.source.TableDeviceQuerySource;
import org.apache.iotdb.db.queryengine.execution.relational.ColumnTransformerBuilder;
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
@@ -52,11 +53,15 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceSchemaCache;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DeleteDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SymbolReference;
import org.apache.iotdb.db.queryengine.transformation.dag.column.ColumnTransformer;
@@ -1525,10 +1530,69 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
throws MetadataException {
if (mtree.deleteTableDevice(
deleteTableDeviceNode.getTableName(), deviceAttributeStore::removeAttribute)) {
+ deviceAttributeCacheUpdater.invalidate(deleteTableDeviceNode.getTableName());
writeToMLog(deleteTableDeviceNode);
}
}
+ @Override
+ public long constructTableDevicesBlackList(
+ final ConstructTableDevicesBlackListNode constructTableDevicesBlackListNode)
+ throws MetadataException {
+ final List<PartialPath> paths =
+ DeleteDevice.constructPaths(
+ PathUtils.unQualifyDatabaseName(storageGroupFullPath),
+ constructTableDevicesBlackListNode.getTableName(),
+ constructTableDevicesBlackListNode.getPatternInfo());
+ final DeviceBlackListConstructor constructor =
+ DeleteDevice.constructDevicePredicateUpdater(
+ PathUtils.unQualifyDatabaseName(storageGroupFullPath),
+ constructTableDevicesBlackListNode.getTableName(),
+ constructTableDevicesBlackListNode.getFilterInfo(),
+ (pointer, name) -> deviceAttributeStore.getAttribute(pointer, name),
+ regionStatistics);
+ try {
+ for (final PartialPath pattern : paths) {
+ mtree.constructTableDeviceBlackList(pattern, constructor);
+ }
+ } finally {
+ constructor.close();
+ }
+ writeToMLog(constructTableDevicesBlackListNode);
+ return constructor.getPreDeletedNum();
+ }
+
+ @Override
+ public void rollbackTableDevicesBlackList(
+ final RollbackTableDevicesBlackListNode rollbackTableDevicesBlackListNode)
+ throws MetadataException {
+ final List<PartialPath> paths =
+ DeleteDevice.constructPaths(
+ PathUtils.unQualifyDatabaseName(storageGroupFullPath),
+ rollbackTableDevicesBlackListNode.getTableName(),
+ rollbackTableDevicesBlackListNode.getPatternInfo());
+ for (final PartialPath pattern : paths) {
+ mtree.rollbackTableDeviceBlackList(pattern);
+ }
+ writeToMLog(rollbackTableDevicesBlackListNode);
+ }
+
+ @Override
+ public void deleteTableDevicesInBlackList(
+ final DeleteTableDevicesInBlackListNode rollbackTableDevicesBlackListNode)
+ throws MetadataException {
+ final List<PartialPath> paths =
+ DeleteDevice.constructPaths(
+ PathUtils.unQualifyDatabaseName(storageGroupFullPath),
+ rollbackTableDevicesBlackListNode.getTableName(),
+ rollbackTableDevicesBlackListNode.getPatternInfo());
+ for (final PartialPath pattern : paths) {
+ mtree.deleteTableDevicesInBlackList(
+ pattern, deviceAttributeStore::removeAttribute, deviceAttributeCacheUpdater::invalidate);
+ }
+ writeToMLog(rollbackTableDevicesBlackListNode);
+ }
+
@Override
public ISchemaReader<IDeviceSchemaInfo> getDeviceReader(final IShowDevicesPlan showDevicesPlan)
throws MetadataException {
@@ -1870,5 +1934,41 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
return new RecoverOperationResult(e);
}
}
+
+ @Override
+ public RecoverOperationResult visitConstructTableDevicesBlackList(
+ final ConstructTableDevicesBlackListNode constructTableDevicesBlackListPlan,
+ final SchemaRegionMemoryImpl context) {
+ try {
+ constructTableDevicesBlackList(constructTableDevicesBlackListPlan);
+ return RecoverOperationResult.SUCCESS;
+ } catch (final MetadataException e) {
+ return new RecoverOperationResult(e);
+ }
+ }
+
+ @Override
+ public RecoverOperationResult visitRollbackTableDevicesBlackList(
+ final RollbackTableDevicesBlackListNode constructTableDevicesBlackListPlan,
+ final SchemaRegionMemoryImpl context) {
+ try {
+ rollbackTableDevicesBlackList(constructTableDevicesBlackListPlan);
+ return RecoverOperationResult.SUCCESS;
+ } catch (final MetadataException e) {
+ return new RecoverOperationResult(e);
+ }
+ }
+
+ @Override
+ public RecoverOperationResult visitDeleteTableDevicesInBlackList(
+ final DeleteTableDevicesInBlackListNode deleteTableDevicesInBlackListPlan,
+ final SchemaRegionMemoryImpl context) {
+ try {
+ deleteTableDevicesInBlackList(deleteTableDevicesInBlackListPlan);
+ return RecoverOperationResult.SUCCESS;
+ } catch (final MetadataException e) {
+ return new RecoverOperationResult(e);
+ }
+ }
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
index 501c183bebf..9a1747969c0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
@@ -39,8 +39,11 @@ import org.apache.iotdb.db.exception.metadata.PathAlreadyExistException;
import org.apache.iotdb.db.exception.metadata.SchemaDirCreationFailureException;
import org.apache.iotdb.db.exception.metadata.SchemaQuotaExceededException;
import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode;
@@ -1450,6 +1453,24 @@ public class SchemaRegionPBTreeImpl implements ISchemaRegion {
throw new UnsupportedOperationException("TableModel does not support PBTree yet.");
}
+ @Override
+ public long constructTableDevicesBlackList(
+ final ConstructTableDevicesBlackListNode constructTableDevicesBlackListNode) {
+ throw new UnsupportedOperationException("TableModel does not support PBTree yet.");
+ }
+
+ @Override
+ public void rollbackTableDevicesBlackList(
+ final RollbackTableDevicesBlackListNode rollbackTableDevicesBlackListNode) {
+ throw new UnsupportedOperationException("TableModel does not support PBTree yet.");
+ }
+
+ @Override
+ public void deleteTableDevicesInBlackList(
+ final DeleteTableDevicesInBlackListNode rollbackTableDevicesBlackListNode) {
+ throw new UnsupportedOperationException("TableModel does not support PBTree yet.");
+ }
+
@Override
public ISchemaReader<IDeviceSchemaInfo> getDeviceReader(IShowDevicesPlan showDevicesPlan)
throws MetadataException {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/logfile/visitor/SchemaRegionPlanDeserializer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/logfile/visitor/SchemaRegionPlanDeserializer.java
index 3062792bb93..b3d27d88336 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/logfile/visitor/SchemaRegionPlanDeserializer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/logfile/visitor/SchemaRegionPlanDeserializer.java
@@ -25,8 +25,11 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathDeserializeUtil;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode;
@@ -441,8 +444,29 @@ public class SchemaRegionPlanDeserializer implements IDeserializer<ISchemaRegion
@Override
public ISchemaRegionPlan visitDeleteTableDevice(
- final DeleteTableDeviceNode addNodeLocationPlan, final ByteBuffer buffer) {
+ final DeleteTableDeviceNode deleteTableDevicePlan, final ByteBuffer buffer) {
return (DeleteTableDeviceNode) PlanNodeType.deserialize(buffer);
}
+
+ @Override
+ public ISchemaRegionPlan visitConstructTableDevicesBlackList(
+ final ConstructTableDevicesBlackListNode constructTableDevicesBlackListPlan,
+ final ByteBuffer buffer) {
+ return (ConstructTableDevicesBlackListNode) PlanNodeType.deserialize(buffer);
+ }
+
+ @Override
+ public ISchemaRegionPlan visitRollbackTableDevicesBlackList(
+ final RollbackTableDevicesBlackListNode rollbackTableDevicesBlackListPlan,
+ final ByteBuffer buffer) {
+ return (RollbackTableDevicesBlackListNode) PlanNodeType.deserialize(buffer);
+ }
+
+ @Override
+ public ISchemaRegionPlan visitDeleteTableDevicesInBlackList(
+ final DeleteTableDevicesInBlackListNode rollbackTableDevicesBlackListPlan,
+ final ByteBuffer buffer) {
+ return (DeleteTableDevicesInBlackListNode) PlanNodeType.deserialize(buffer);
+ }
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/logfile/visitor/SchemaRegionPlanSerializer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/logfile/visitor/SchemaRegionPlanSerializer.java
index 128b20f17be..c344cffbbc2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/logfile/visitor/SchemaRegionPlanSerializer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/logfile/visitor/SchemaRegionPlanSerializer.java
@@ -22,8 +22,11 @@ package org.apache.iotdb.db.schemaengine.schemaregion.logfile.visitor;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode;
@@ -513,6 +516,27 @@ public class SchemaRegionPlanSerializer implements ISerializer<ISchemaRegionPlan
return visitPlanNode(deleteTableDevicePlan, outputStream);
}
+ @Override
+ public SchemaRegionPlanSerializationResult visitConstructTableDevicesBlackList(
+ final ConstructTableDevicesBlackListNode constructTableDevicesBlackListPlan,
+ final DataOutputStream outputStream) {
+ return visitPlanNode(constructTableDevicesBlackListPlan, outputStream);
+ }
+
+ @Override
+ public SchemaRegionPlanSerializationResult visitRollbackTableDevicesBlackList(
+ final RollbackTableDevicesBlackListNode rollbackTableDevicesBlackListPlan,
+ final DataOutputStream outputStream) {
+ return visitPlanNode(rollbackTableDevicesBlackListPlan, outputStream);
+ }
+
+ @Override
+ public SchemaRegionPlanSerializationResult visitDeleteTableDevicesInBlackList(
+ final DeleteTableDevicesInBlackListNode deleteTableDevicesInBlackListPlan,
+ final DataOutputStream outputStream) {
+ return visitPlanNode(deleteTableDevicesInBlackListPlan, outputStream);
+ }
+
private SchemaRegionPlanSerializationResult visitPlanNode(
final PlanNode planNode, final DataOutputStream outputStream) {
try {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java
index 4fe9dd75686..5ada2ffedda 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java
@@ -45,6 +45,7 @@ import org.apache.iotdb.db.exception.metadata.template.TemplateIsInUseException;
import org.apache.iotdb.db.exception.quota.ExceedQuotaException;
import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
import org.apache.iotdb.db.queryengine.execution.operator.schema.source.DeviceAttributeUpdater;
+import org.apache.iotdb.db.queryengine.execution.operator.schema.source.DeviceBlackListConstructor;
import org.apache.iotdb.db.schemaengine.metric.SchemaRegionMemMetric;
import org.apache.iotdb.db.schemaengine.rescon.MemSchemaRegionStatistics;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.IMemMNode;
@@ -638,7 +639,7 @@ public class MTreeBelowSGMemoryImpl {
try (final MeasurementUpdater<IMemMNode> updater =
new MeasurementUpdater<IMemMNode>(
rootNode, pathPattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
-
+ @Override
protected void updateMeasurement(final IMeasurementMNode<IMemMNode> node) {
if (!node.isLogicalView()) {
isAllLogicalView.set(false);
@@ -658,7 +659,7 @@ public class MTreeBelowSGMemoryImpl {
try (final MeasurementUpdater<IMemMNode> updater =
new MeasurementUpdater<IMemMNode>(
rootNode, pathPattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
-
+ @Override
protected void updateMeasurement(final IMeasurementMNode<IMemMNode> node) {
node.setPreDeleted(false);
result.add(getPartialPathFromRootToNode(node.getAsMNode()));
@@ -675,7 +676,7 @@ public class MTreeBelowSGMemoryImpl {
try (final MeasurementCollector<Void, IMemMNode> collector =
new MeasurementCollector<Void, IMemMNode>(
rootNode, pathPattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
-
+ @Override
protected Void collectMeasurement(final IMeasurementMNode<IMemMNode> node) {
if (node.isPreDeleted()) {
result.add(getPartialPathFromRootToNode(node.getAsMNode()));
@@ -695,7 +696,7 @@ public class MTreeBelowSGMemoryImpl {
try (final MeasurementCollector<Void, IMemMNode> collector =
new MeasurementCollector<Void, IMemMNode>(
rootNode, pathPattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
-
+ @Override
protected Void collectMeasurement(final IMeasurementMNode<IMemMNode> node) {
if (node.isPreDeleted()) {
result.add(getPartialPathFromRootToNode(node.getAsMNode()).getDevicePath());
@@ -770,6 +771,7 @@ public class MTreeBelowSGMemoryImpl {
try (final MeasurementCollector<Void, IMemMNode> collector =
new MeasurementCollector<Void, IMemMNode>(
rootNode, pathPattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
+ @Override
protected Void collectMeasurement(IMeasurementMNode<IMemMNode> node) {
final IDeviceMNode<IMemMNode> deviceMNode =
getParentOfNextMatchedNode().getAsDeviceMNode();
@@ -811,6 +813,7 @@ public class MTreeBelowSGMemoryImpl {
try (final MeasurementCollector<Void, IMemMNode> collector =
new MeasurementCollector<Void, IMemMNode>(
rootNode, patternTree, store, SchemaConstant.ALL_MATCH_SCOPE) {
+ @Override
protected Void collectMeasurement(IMeasurementMNode<IMemMNode> node) {
final IDeviceMNode<IMemMNode> deviceMNode =
getParentOfNextMatchedNode().getAsDeviceMNode();
@@ -946,11 +949,12 @@ public class MTreeBelowSGMemoryImpl {
new EntityUpdater<IMemMNode>(
rootNode, entry.getKey(), store, false, SchemaConstant.ALL_MATCH_SCOPE) {
+ @Override
protected void updateEntity(final IDeviceMNode<IMemMNode> node) {
if (entry.getValue().contains(node.getSchemaTemplateId())) {
resultTemplateSetInfo.put(
node.getPartialPath(), Collections.singletonList(node.getSchemaTemplateId()));
- node.preDeactivateTemplate();
+ node.preDeactivateSelfOrTemplate();
}
}
}) {
@@ -968,12 +972,13 @@ public class MTreeBelowSGMemoryImpl {
new EntityUpdater<IMemMNode>(
rootNode, entry.getKey(), store, false, SchemaConstant.ALL_MATCH_SCOPE) {
+ @Override
protected void updateEntity(final IDeviceMNode<IMemMNode> node) {
if (entry.getValue().contains(node.getSchemaTemplateId())
- && node.isPreDeactivateTemplate()) {
+ && node.isPreDeactivateSelfOrTemplate()) {
resultTemplateSetInfo.put(
node.getPartialPath(), Collections.singletonList(node.getSchemaTemplateId()));
- node.rollbackPreDeactivateTemplate();
+ node.rollbackPreDeactivateSelfOrTemplate();
}
}
}) {
@@ -991,9 +996,10 @@ public class MTreeBelowSGMemoryImpl {
new EntityUpdater<IMemMNode>(
rootNode, entry.getKey(), store, false, SchemaConstant.ALL_MATCH_SCOPE) {
+ @Override
protected void updateEntity(final IDeviceMNode<IMemMNode> node) {
if (entry.getValue().contains(node.getSchemaTemplateId())
- && node.isPreDeactivateTemplate()) {
+ && node.isPreDeactivateSelfOrTemplate()) {
resultTemplateSetInfo.put(
node.getPartialPath(), Collections.singletonList(node.getSchemaTemplateId()));
regionStatistics.deactivateTemplate(node.getSchemaTemplateId());
@@ -1057,6 +1063,7 @@ public class MTreeBelowSGMemoryImpl {
showDevicesPlan.isPrefixMatch(),
showDevicesPlan.getScope()) {
+ @Override
protected IDeviceSchemaInfo collectEntity(final IDeviceMNode<IMemMNode> node) {
final PartialPath device = getPartialPathFromRootToNode(node.getAsMNode());
final ShowDevicesResult result =
@@ -1079,22 +1086,27 @@ public class MTreeBelowSGMemoryImpl {
private final DeviceFilterVisitor filterVisitor = new DeviceFilterVisitor();
private IDeviceSchemaInfo next;
+ @Override
public boolean isSuccess() {
return collector.isSuccess();
}
+ @Override
public Throwable getFailure() {
return collector.getFailure();
}
+ @Override
public void close() {
collector.close();
}
+ @Override
public ListenableFuture<?> isBlocked() {
return NOT_BLOCKED;
}
+ @Override
public boolean hasNext() {
while (next == null && collector.hasNext()) {
IDeviceSchemaInfo temp = collector.next();
@@ -1106,6 +1118,7 @@ public class MTreeBelowSGMemoryImpl {
return next != null;
}
+ @Override
public IDeviceSchemaInfo next() {
if (!hasNext()) {
throw new NoSuchElementException();
@@ -1130,6 +1143,12 @@ public class MTreeBelowSGMemoryImpl {
final EntityCollector<IDeviceSchemaInfo, IMemMNode> collector =
new EntityCollector<IDeviceSchemaInfo, IMemMNode>(rootNode, pattern, store, false, null) {
+ @Override
+ protected boolean acceptFullMatchedNode(final IMemMNode node) {
+ return node.isDevice() && !node.getAsDeviceMNode().isPreDeactivateSelfOrTemplate();
+ }
+
+ @Override
protected IDeviceSchemaInfo collectEntity(final IDeviceMNode<IMemMNode> node) {
final ShowDevicesResult result =
new ShowDevicesResult(
@@ -1147,26 +1166,32 @@ public class MTreeBelowSGMemoryImpl {
};
return new ISchemaReader<IDeviceSchemaInfo>() {
+ @Override
public boolean isSuccess() {
return collector.isSuccess();
}
+ @Override
public Throwable getFailure() {
return collector.getFailure();
}
+ @Override
public void close() {
collector.close();
}
+ @Override
public ListenableFuture<?> isBlocked() {
return NOT_BLOCKED;
}
+ @Override
public boolean hasNext() {
return collector.hasNext();
}
+ @Override
public IDeviceSchemaInfo next() {
return collector.next();
}
@@ -1224,7 +1249,7 @@ public class MTreeBelowSGMemoryImpl {
try {
final IMemMNode node = getTableDeviceNode(table, deviceIdIterator.next());
- if (!node.isDevice()) {
+ if (!node.isDevice() || node.getAsDeviceMNode().isPreDeactivateSelfOrTemplate()) {
continue;
}
final IDeviceMNode<IMemMNode> deviceNode = node.getAsDeviceMNode();
@@ -1305,20 +1330,24 @@ public class MTreeBelowSGMemoryImpl {
showTimeSeriesPlan.isPrefixMatch(),
showTimeSeriesPlan.getScope()) {
+ @Override
protected ITimeSeriesSchemaInfo collectMeasurement(
final IMeasurementMNode<IMemMNode> node) {
return new ITimeSeriesSchemaInfo() {
private Pair<Map<String, String>, Map<String, String>> tagAndAttribute = null;
+ @Override
public String getAlias() {
return node.getAlias();
}
+ @Override
public IMeasurementSchema getSchema() {
return node.getSchema();
}
+ @Override
public Map<String, String> getTags() {
if (tagAndAttribute == null) {
tagAndAttribute = tagAndAttributeProvider.apply(node.getOffset());
@@ -1326,6 +1355,7 @@ public class MTreeBelowSGMemoryImpl {
return tagAndAttribute.left;
}
+ @Override
public Map<String, String> getAttributes() {
if (tagAndAttribute == null) {
tagAndAttribute = tagAndAttributeProvider.apply(node.getOffset());
@@ -1333,6 +1363,7 @@ public class MTreeBelowSGMemoryImpl {
return tagAndAttribute.right;
}
+ @Override
public boolean isUnderAlignedDevice() {
return getParentOfNextMatchedNode().getAsDeviceMNode().isAligned();
}
@@ -1342,10 +1373,12 @@ public class MTreeBelowSGMemoryImpl {
return node.isLogicalView();
}
+ @Override
public String getFullPath() {
return getPartialPathFromRootToNode(node.getAsMNode()).getFullPath();
}
+ @Override
public PartialPath getPartialPath() {
return getPartialPathFromRootToNode(node.getAsMNode());
}
@@ -1382,6 +1415,7 @@ public class MTreeBelowSGMemoryImpl {
showNodesPlan.isPrefixMatch(),
showNodesPlan.getScope()) {
+ @Override
protected INodeSchemaInfo collectMNode(final IMemMNode node) {
return new ShowNodesResult(
getPartialPathFromRootToNode(node).getFullPath(), node.getMNodeType());
@@ -1391,26 +1425,32 @@ public class MTreeBelowSGMemoryImpl {
return new ISchemaReader<INodeSchemaInfo>() {
+ @Override
public boolean isSuccess() {
return collector.isSuccess();
}
+ @Override
public Throwable getFailure() {
return collector.getFailure();
}
+ @Override
public void close() {
collector.close();
}
+ @Override
public ListenableFuture<?> isBlocked() {
return NOT_BLOCKED;
}
+ @Override
public boolean hasNext() {
return collector.hasNext();
}
+ @Override
public INodeSchemaInfo next() {
if (!hasNext()) {
throw new NoSuchElementException();
@@ -1480,7 +1520,7 @@ public class MTreeBelowSGMemoryImpl {
try (final MeasurementUpdater<IMemMNode> updater =
new MeasurementUpdater<IMemMNode>(
rootNode, pathPattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
-
+ @Override
protected void updateMeasurement(final IMeasurementMNode<IMemMNode> node) {
if (node.isLogicalView()) {
node.setPreDeleted(true);
@@ -1499,7 +1539,7 @@ public class MTreeBelowSGMemoryImpl {
try (final MeasurementUpdater<IMemMNode> updater =
new MeasurementUpdater<IMemMNode>(
rootNode, pathPattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
-
+ @Override
protected void updateMeasurement(final IMeasurementMNode<IMemMNode> node) {
if (node.isLogicalView()) {
node.setPreDeleted(false);
@@ -1519,6 +1559,7 @@ public class MTreeBelowSGMemoryImpl {
new MeasurementCollector<Void, IMemMNode>(
rootNode, pathPattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
+ @Override
protected Void collectMeasurement(final IMeasurementMNode<IMemMNode> node) {
if (node.isLogicalView() && node.isPreDeleted()) {
result.add(getPartialPathFromRootToNode(node.getAsMNode()));
@@ -1595,6 +1636,63 @@ public class MTreeBelowSGMemoryImpl {
}
}
+ public void constructTableDeviceBlackList(
+ final PartialPath pattern, final DeviceBlackListConstructor blackListConstructor)
+ throws MetadataException {
+ try (final EntityUpdater<IMemMNode> updater =
+ new EntityUpdater<IMemMNode>(
+ rootNode, pattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
+
+ @Override
+ protected void updateEntity(final IDeviceMNode<IMemMNode> node) {
+ blackListConstructor.handleDeviceNode(node);
+ }
+ }) {
+ updater.update();
+ }
+ }
+
+ public void rollbackTableDeviceBlackList(final PartialPath pattern) throws MetadataException {
+ try (final EntityUpdater<IMemMNode> updater =
+ new EntityUpdater<IMemMNode>(
+ rootNode, pattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
+
+ @Override
+ protected void updateEntity(final IDeviceMNode<IMemMNode> node) {
+ if (node.isPreDeactivateSelfOrTemplate()) {
+ regionStatistics.addTableDevice(pattern.getNodes()[2]);
+ node.rollbackPreDeactivateSelfOrTemplate();
+ }
+ }
+ }) {
+ updater.update();
+ }
+ }
+
+ public void deleteTableDevicesInBlackList(
+ final PartialPath pattern,
+ final IntConsumer attributeDeleter,
+ final Consumer<String[]> deviceAttributeCacheUpdateInvalidator)
+ throws MetadataException {
+ try (final EntityUpdater<IMemMNode> updater =
+ new EntityUpdater<IMemMNode>(
+ rootNode, pattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) {
+
+ @Override
+ protected void updateEntity(final IDeviceMNode<IMemMNode> node) {
+ if (node.isPreDeactivateSelfOrTemplate()) {
+ attributeDeleter.accept(
+ ((TableDeviceInfo<IMemMNode>) node.getAsDeviceMNode().getDeviceInfo())
+ .getAttributePointer());
+ deviceAttributeCacheUpdateInvalidator.accept(node.getPartialPath().getNodes());
+ deleteEmptyInternalMNode(node);
+ }
+ }
+ }) {
+ updater.update();
+ }
+ }
+
public void renameTableAttribute() {}
public boolean deleteTableDevice(final String tableName, final IntConsumer attributeDeleter)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/factory/MemMNodeFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/factory/MemMNodeFactory.java
index 0850f6539d9..19e24ed87b9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/factory/MemMNodeFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/factory/MemMNodeFactory.java
@@ -28,7 +28,7 @@ import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.impl.A
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.impl.BasicInternalMNode;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.impl.DatabaseMNode;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.impl.MeasurementMNode;
-import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.DeviceInfo;
+import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.TreeDeviceInfo;
import org.apache.tsfile.write.schema.IMeasurementSchema;
@@ -44,7 +44,7 @@ public class MemMNodeFactory implements IMNodeFactory<IMemMNode> {
@Override
public IDeviceMNode<IMemMNode> createDeviceMNode(IMemMNode parent, String name) {
BasicInternalMNode internalMNode = new BasicInternalMNode(parent, name);
- internalMNode.setDeviceInfo(new DeviceInfo<>());
+ internalMNode.setDeviceInfo(new TreeDeviceInfo<>());
return internalMNode.getAsDeviceMNode();
}
@@ -56,7 +56,7 @@ public class MemMNodeFactory implements IMNodeFactory<IMemMNode> {
@Override
public IMemMNode createDatabaseDeviceMNode(IMemMNode parent, String name) {
DatabaseMNode databaseMNode = new DatabaseMNode(parent, name);
- databaseMNode.setDeviceInfo(new DeviceInfo<>());
+ databaseMNode.setDeviceInfo(new TreeDeviceInfo<>());
return databaseMNode.getAsMNode();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/DatabaseDeviceInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/DatabaseDeviceInfo.java
index dde32311a48..304d9e161a0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/DatabaseDeviceInfo.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/DatabaseDeviceInfo.java
@@ -22,7 +22,7 @@ import org.apache.iotdb.commons.schema.node.IMNode;
import org.apache.iotdb.commons.schema.node.info.IDatabaseDeviceInfo;
import org.apache.iotdb.commons.schema.node.role.IDatabaseMNode;
-public class DatabaseDeviceInfo<N extends IMNode<N>> extends DeviceInfo<N>
+public class DatabaseDeviceInfo<N extends IMNode<N>> extends TreeDeviceInfo<N>
implements IDatabaseDeviceInfo<N> {
/**
* when the data file in a database is older than dataTTL, it is considered invalid and will be
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/TableDeviceInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/TableDeviceInfo.java
index 857cf20a40d..f3271435bc9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/TableDeviceInfo.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/TableDeviceInfo.java
@@ -33,23 +33,23 @@ public class TableDeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
private int attributePointer = -1;
public int getAttributePointer() {
- return attributePointer;
+ return attributePointer >= -1 ? attributePointer : -attributePointer - 2;
}
- public void setAttributePointer(int attributePointer) {
+ public void setAttributePointer(final int attributePointer) {
this.attributePointer = attributePointer;
}
@Override
- public void moveDataToNewMNode(IDeviceMNode<N> newMNode) {}
+ public void moveDataToNewMNode(final IDeviceMNode<N> newMNode) {}
@Override
- public boolean addAlias(String alias, IMeasurementMNode<N> child) {
+ public boolean addAlias(final String alias, final IMeasurementMNode<N> child) {
return false;
}
@Override
- public void deleteAliasChild(String alias) {}
+ public void deleteAliasChild(final String alias) {}
@Override
public Map<String, IMeasurementMNode<N>> getAliasChildren() {
@@ -57,15 +57,15 @@ public class TableDeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
}
@Override
- public void setAliasChildren(Map<String, IMeasurementMNode<N>> aliasChildren) {}
+ public void setAliasChildren(final Map<String, IMeasurementMNode<N>> aliasChildren) {}
@Override
- public boolean hasAliasChild(String name) {
+ public boolean hasAliasChild(final String name) {
return false;
}
@Override
- public N getAliasChild(String name) {
+ public N getAliasChild(final String name) {
return null;
}
@@ -75,10 +75,10 @@ public class TableDeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
}
@Override
- public void setUseTemplate(boolean useTemplate) {}
+ public void setUseTemplate(final boolean useTemplate) {}
@Override
- public void setSchemaTemplateId(int schemaTemplateId) {}
+ public void setSchemaTemplateId(final int schemaTemplateId) {}
@Override
public int getSchemaTemplateId() {
@@ -91,15 +91,23 @@ public class TableDeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
}
@Override
- public boolean isPreDeactivateTemplate() {
- return false;
+ public boolean isPreDeactivateSelfOrTemplate() {
+ return attributePointer < -1;
}
@Override
- public void preDeactivateTemplate() {}
+ public void preDeactivateSelfOrTemplate() {
+ if (attributePointer > -1) {
+ attributePointer = -attributePointer - 2;
+ }
+ }
@Override
- public void rollbackPreDeactivateTemplate() {}
+ public void rollbackPreDeactivateSelfOrTemplate() {
+ if (attributePointer < -1) {
+ attributePointer = -attributePointer - 2;
+ }
+ }
@Override
public void deactivateTemplate() {}
@@ -110,7 +118,7 @@ public class TableDeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
}
@Override
- public void setAligned(Boolean isAligned) {}
+ public void setAligned(final Boolean isAligned) {}
@Override
public int estimateSize() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/DeviceInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/TreeDeviceInfo.java
similarity index 86%
rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/DeviceInfo.java
rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/TreeDeviceInfo.java
index 81d4015bd35..1fc8269a54b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/DeviceInfo.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/info/TreeDeviceInfo.java
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info;
import org.apache.iotdb.commons.schema.node.IMNode;
@@ -29,7 +30,7 @@ import java.util.concurrent.ConcurrentHashMap;
import static org.apache.iotdb.commons.schema.SchemaConstant.NON_TEMPLATE;
-public class DeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
+public class TreeDeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
/**
* In EntityMNode of MTree in SchemaRegion, this field represents the template activated on this
@@ -52,7 +53,7 @@ public class DeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
private volatile Boolean isAligned = false;
@Override
- public void moveDataToNewMNode(IDeviceMNode<N> newMNode) {
+ public void moveDataToNewMNode(final IDeviceMNode<N> newMNode) {
newMNode.setSchemaTemplateId(schemaTemplateId);
newMNode.setUseTemplate(useTemplate);
newMNode.setAliasChildren(aliasChildren);
@@ -61,7 +62,7 @@ public class DeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
/** add an alias */
@Override
- public boolean addAlias(String alias, IMeasurementMNode<N> child) {
+ public boolean addAlias(final String alias, final IMeasurementMNode<N> child) {
if (aliasChildren == null) {
// double check, alias children volatile
synchronized (this) {
@@ -76,7 +77,7 @@ public class DeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
/** delete the alias of a child */
@Override
- public void deleteAliasChild(String alias) {
+ public void deleteAliasChild(final String alias) {
if (aliasChildren != null) {
aliasChildren.remove(alias);
}
@@ -91,17 +92,17 @@ public class DeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
}
@Override
- public void setAliasChildren(Map<String, IMeasurementMNode<N>> aliasChildren) {
+ public void setAliasChildren(final Map<String, IMeasurementMNode<N>> aliasChildren) {
this.aliasChildren = aliasChildren;
}
@Override
- public boolean hasAliasChild(String name) {
+ public boolean hasAliasChild(final String name) {
return aliasChildren != null && aliasChildren.containsKey(name);
}
@Override
- public N getAliasChild(String name) {
+ public N getAliasChild(final String name) {
if (aliasChildren != null) {
IMeasurementMNode<N> child = aliasChildren.get(name);
return child == null ? null : child.getAsMNode();
@@ -128,19 +129,19 @@ public class DeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
}
@Override
- public boolean isPreDeactivateTemplate() {
+ public boolean isPreDeactivateSelfOrTemplate() {
return schemaTemplateId < -1;
}
@Override
- public void preDeactivateTemplate() {
+ public void preDeactivateSelfOrTemplate() {
if (schemaTemplateId > -1) {
schemaTemplateId = -schemaTemplateId - 2;
}
}
@Override
- public void rollbackPreDeactivateTemplate() {
+ public void rollbackPreDeactivateSelfOrTemplate() {
if (schemaTemplateId < -1) {
schemaTemplateId = -schemaTemplateId - 2;
}
@@ -158,12 +159,12 @@ public class DeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
}
@Override
- public void setUseTemplate(boolean useTemplate) {
+ public void setUseTemplate(final boolean useTemplate) {
this.useTemplate = useTemplate;
}
@Override
- public void setSchemaTemplateId(int schemaTemplateId) {
+ public void setSchemaTemplateId(final int schemaTemplateId) {
this.schemaTemplateId = schemaTemplateId;
}
@@ -173,7 +174,7 @@ public class DeviceInfo<N extends IMNode<N>> implements IDeviceInfo<N> {
}
@Override
- public void setAligned(Boolean isAligned) {
+ public void setAligned(final Boolean isAligned) {
this.isAligned = isAligned;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/iterator/AbstractTraverserIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/iterator/AbstractTraverserIterator.java
index f7f6d7d1ae6..6f95102a26f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/iterator/AbstractTraverserIterator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/iterator/AbstractTraverserIterator.java
@@ -76,7 +76,7 @@ public abstract class AbstractTraverserIterator<N extends IMNode<N>> implements
IDeviceMNode<N> node, Map<Integer, Template> templateMap) {
// new cluster, the used template is directly recorded as template id in device mnode
if (node.getSchemaTemplateId() != NON_TEMPLATE) {
- if (skipPreDeletedSchema && node.getAsDeviceMNode().isPreDeactivateTemplate()) {
+ if (skipPreDeletedSchema && node.getAsDeviceMNode().isPreDeactivateSelfOrTemplate()) {
// skip this pre deactivated template, the invoker will skip this
return null;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java
index 332159833d7..b6b1a0777be 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java
@@ -1260,7 +1260,7 @@ public class MTreeBelowSGCachedImpl {
resultTemplateSetInfo.put(
node.getPartialPath(), Collections.singletonList(node.getSchemaTemplateId()));
store.updateMNode(
- node.getAsMNode(), o -> o.getAsDeviceMNode().preDeactivateTemplate());
+ node.getAsMNode(), o -> o.getAsDeviceMNode().preDeactivateSelfOrTemplate());
}
}
}) {
@@ -1280,11 +1280,12 @@ public class MTreeBelowSGCachedImpl {
protected void updateEntity(IDeviceMNode<ICachedMNode> node) throws MetadataException {
if (entry.getValue().contains(node.getSchemaTemplateId())
- && node.isPreDeactivateTemplate()) {
+ && node.isPreDeactivateSelfOrTemplate()) {
resultTemplateSetInfo.put(
node.getPartialPath(), Collections.singletonList(node.getSchemaTemplateId()));
store.updateMNode(
- node.getAsMNode(), o -> o.getAsDeviceMNode().rollbackPreDeactivateTemplate());
+ node.getAsMNode(),
+ o -> o.getAsDeviceMNode().rollbackPreDeactivateSelfOrTemplate());
}
}
}) {
@@ -1304,7 +1305,7 @@ public class MTreeBelowSGCachedImpl {
protected void updateEntity(IDeviceMNode<ICachedMNode> node) throws MetadataException {
if (entry.getValue().contains(node.getSchemaTemplateId())
- && node.isPreDeactivateTemplate()) {
+ && node.isPreDeactivateSelfOrTemplate()) {
resultTemplateSetInfo.put(
node.getPartialPath(), Collections.singletonList(node.getSchemaTemplateId()));
regionStatistics.deactivateTemplate(node.getSchemaTemplateId());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/mnode/factory/CacheMNodeFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/mnode/factory/CacheMNodeFactory.java
index bc51c722f67..92da4db1948 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/mnode/factory/CacheMNodeFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/mnode/factory/CacheMNodeFactory.java
@@ -23,7 +23,7 @@ import org.apache.iotdb.commons.schema.node.role.IDeviceMNode;
import org.apache.iotdb.commons.schema.node.role.IMeasurementMNode;
import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory;
import org.apache.iotdb.commons.schema.node.utils.MNodeFactory;
-import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.DeviceInfo;
+import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.TreeDeviceInfo;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.ICachedMNode;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.impl.CachedAboveDatabaseMNode;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.impl.CachedBasicInternalMNode;
@@ -44,7 +44,7 @@ public class CacheMNodeFactory implements IMNodeFactory<ICachedMNode> {
@Override
public IDeviceMNode<ICachedMNode> createDeviceMNode(ICachedMNode parent, String name) {
CachedBasicInternalMNode internalMNode = new CachedBasicInternalMNode(parent, name);
- internalMNode.setDeviceInfo(new DeviceInfo<>());
+ internalMNode.setDeviceInfo(new TreeDeviceInfo<>());
return internalMNode.getAsDeviceMNode();
}
@@ -56,7 +56,7 @@ public class CacheMNodeFactory implements IMNodeFactory<ICachedMNode> {
@Override
public ICachedMNode createDatabaseDeviceMNode(ICachedMNode parent, String name) {
CachedDatabaseMNode databaseMNode = new CachedDatabaseMNode(parent, name);
- databaseMNode.setDeviceInfo(new DeviceInfo<>());
+ databaseMNode.setDeviceInfo(new TreeDeviceInfo<>());
return databaseMNode;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/traverser/Traverser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/traverser/Traverser.java
index fb7294965a9..211f9e19999 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/traverser/Traverser.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/traverser/Traverser.java
@@ -163,7 +163,7 @@ public abstract class Traverser<R, N extends IMNode<N>> extends AbstractTreeVisi
&& !(skipPreDeletedSchema
&& parent
.getAsDeviceMNode()
- .isPreDeactivateTemplate())) { // the template should not skip
+ .isPreDeactivateSelfOrTemplate())) { // the template should not skip
int templateId = parent.getAsDeviceMNode().getSchemaTemplateId();
Template template = templateMap.get(templateId);
// If null, it means the template on this device is not covered in this query, refer to the
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/MNodeUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/MNodeUtils.java
index 3506f6f9ba8..b896e6e0d9d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/MNodeUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/MNodeUtils.java
@@ -22,7 +22,7 @@ import org.apache.iotdb.commons.schema.node.IMNode;
import org.apache.iotdb.commons.schema.node.role.IDeviceMNode;
import org.apache.iotdb.commons.schema.node.role.IInternalMNode;
import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory;
-import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.DeviceInfo;
+import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.TreeDeviceInfo;
import org.apache.iotdb.db.schemaengine.template.Template;
import org.apache.tsfile.write.schema.IMeasurementSchema;
@@ -41,7 +41,7 @@ public class MNodeUtils {
public static <N extends IMNode<N>> boolean setToEntity(IMNode<N> node) {
IInternalMNode<N> internalMNode = node.getAsInternalMNode();
if (!internalMNode.isDevice()) {
- internalMNode.setDeviceInfo(new DeviceInfo<>());
+ internalMNode.setDeviceInfo(new TreeDeviceInfo<>());
return true;
} else {
return false;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/write/req/SchemaRegionWritePlanFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/write/req/SchemaRegionWritePlanFactory.java
index 41308332579..4696378e447 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/write/req/SchemaRegionWritePlanFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/write/req/SchemaRegionWritePlanFactory.java
@@ -22,8 +22,11 @@ package org.apache.iotdb.db.schemaengine.schemaregion.write.req;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.DeleteTableDevicesInBlackListNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.RollbackTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode;
@@ -109,6 +112,12 @@ public class SchemaRegionWritePlanFactory {
return TableNodeLocationAddNode.MOCK_INSTANCE;
case DELETE_TABLE_DEVICE:
return DeleteTableDeviceNode.MOCK_INSTANCE;
+ case CONSTRUCT_TABLE_DEVICES_BLACK_LIST:
+ return ConstructTableDevicesBlackListNode.MOCK_INSTANCE;
+ case ROLLBACK_TABLE_DEVICES_BLACK_LIST:
+ return RollbackTableDevicesBlackListNode.MOCK_INSTANCE;
+ case DELETE_TABLE_DEVICES_IN_BLACK_LIST:
+ return DeleteTableDevicesInBlackListNode.MOCK_INSTANCE;
default:
throw new UnsupportedOperationException(
String.format(
@@ -117,7 +126,7 @@ public class SchemaRegionWritePlanFactory {
}
}
- public static IChangeAliasPlan getChangeAliasPlan(PartialPath path, String alias) {
+ public static IChangeAliasPlan getChangeAliasPlan(final PartialPath path, final String alias) {
return new ChangeAliasPlanImpl(path, alias);
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
index 04a7570a8ee..0d454b0d640 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
@@ -161,6 +161,7 @@ public class IoTDBConstant {
public static final String COLUMN_CHILD_PATHS_TYPES = "node types";
public static final String COLUMN_CHILD_NODES = "child nodes";
public static final String COLUMN_DEVICES = "devices";
+ public static final String COLUMN_DELETED_DEVICE_NUM = "num_of_deleted_devices";
public static final String COLUMN_COLUMN = "column";
public static final String COLUMN_COUNT = "count";
public static final String COLUMN_TAGS = "tags";
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/common/DeviceMNodeWrapper.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/common/DeviceMNodeWrapper.java
index 6b400832da5..5c564a6fb26 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/common/DeviceMNodeWrapper.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/common/DeviceMNodeWrapper.java
@@ -193,12 +193,12 @@ public class DeviceMNodeWrapper<N extends IMNode<N>, BasicNode extends IInternal
}
@Override
- public void setUseTemplate(boolean useTemplate) {
+ public void setUseTemplate(final boolean useTemplate) {
basicMNode.getDeviceInfo().setUseTemplate(useTemplate);
}
@Override
- public void setSchemaTemplateId(int schemaTemplateId) {
+ public void setSchemaTemplateId(final int schemaTemplateId) {
basicMNode.getDeviceInfo().setSchemaTemplateId(schemaTemplateId);
}
@@ -213,18 +213,18 @@ public class DeviceMNodeWrapper<N extends IMNode<N>, BasicNode extends IInternal
}
@Override
- public boolean isPreDeactivateTemplate() {
- return basicMNode.getDeviceInfo().isPreDeactivateTemplate();
+ public boolean isPreDeactivateSelfOrTemplate() {
+ return basicMNode.getDeviceInfo().isPreDeactivateSelfOrTemplate();
}
@Override
- public void preDeactivateTemplate() {
- basicMNode.getDeviceInfo().preDeactivateTemplate();
+ public void preDeactivateSelfOrTemplate() {
+ basicMNode.getDeviceInfo().preDeactivateSelfOrTemplate();
}
@Override
- public void rollbackPreDeactivateTemplate() {
- basicMNode.getDeviceInfo().rollbackPreDeactivateTemplate();
+ public void rollbackPreDeactivateSelfOrTemplate() {
+ basicMNode.getDeviceInfo().rollbackPreDeactivateSelfOrTemplate();
}
@Override
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/info/IDeviceInfo.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/info/IDeviceInfo.java
index 456a51753e5..d11acc9953e 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/info/IDeviceInfo.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/info/IDeviceInfo.java
@@ -56,11 +56,11 @@ public interface IDeviceInfo<N extends IMNode<N>> {
*/
int getSchemaTemplateIdWithState();
- boolean isPreDeactivateTemplate();
+ boolean isPreDeactivateSelfOrTemplate();
- void preDeactivateTemplate();
+ void preDeactivateSelfOrTemplate();
- void rollbackPreDeactivateTemplate();
+ void rollbackPreDeactivateSelfOrTemplate();
void deactivateTemplate();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/role/IDeviceMNode.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/role/IDeviceMNode.java
index 1df1c51dcfe..99948a09861 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/role/IDeviceMNode.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/node/role/IDeviceMNode.java
@@ -47,11 +47,11 @@ public interface IDeviceMNode<N extends IMNode<N>> extends IInternalMNode<N> {
*/
int getSchemaTemplateIdWithState();
- boolean isPreDeactivateTemplate();
+ boolean isPreDeactivateSelfOrTemplate();
- void preDeactivateTemplate();
+ void preDeactivateSelfOrTemplate();
- void rollbackPreDeactivateTemplate();
+ void rollbackPreDeactivateSelfOrTemplate();
void deactivateTemplate();
diff --git a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
index f9a5908fe83..58d5de183cf 100644
--- a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
+++ b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
@@ -64,6 +64,7 @@ statement
| insertStatement
| updateStatement
| deleteStatement
+ | deleteDeviceStatement
// UDF Statement
| showFunctionsStatement
@@ -220,7 +221,9 @@ updateStatement
: UPDATE qualifiedName SET updateAssignment (',' updateAssignment)* (WHERE where=booleanExpression)?
;
-
+deleteDeviceStatement
+ : DELETE DEVICES FROM tableName=qualifiedName (WHERE booleanExpression)?
+ ;
// -------------------------------------------- UDF Statement ----------------------------------------------------------
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index 4269bb4e671..a535470516f 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -1042,6 +1042,19 @@ struct TAlterOrDropTableReq {
5: required binary updateInfo
}
+struct TDeleteTableDeviceReq {
+ 1: required string database
+ 2: required string tableName
+ 3: required string queryId
+ 4: required binary patternInfo
+ 5: required binary filterInfo
+}
+
+struct TDeleteTableDeviceResp {
+ 1: required common.TSStatus status
+ 2: optional i64 deletedNum
+}
+
struct TShowTableResp {
1: required common.TSStatus status
2: optional list<TTableInfo> tableInfoList
@@ -1796,5 +1809,7 @@ service IConfigNodeRPCService {
TShowTableResp showTables(string database, bool isDetails)
TFetchTableResp fetchTables(map<string, set<string>> fetchTableMap)
+
+ TDeleteTableDeviceResp deleteDevice(TDeleteTableDeviceReq req)
}
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index 32168aff86d..8cc329a1a77 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -331,7 +331,7 @@ struct TUpdateTableReq {
}
struct TInvalidateTableCacheReq {
- 1: required string database,
+ 1: required string database
2: required string tableName
}
@@ -340,6 +340,25 @@ struct TDeleteDataOrDevicesForDropTableReq {
2: required string tableName
}
+struct TTableDeviceDeletionWithPatternAndFilterReq {
+ 1: required list<common.TConsensusGroupId> regionIdList
+ 2: required string tableName
+ 3: required binary patternInfo
+ 4: required binary filterInfo
+}
+
+struct TTableDeviceDeletionWithPatternReq {
+ 1: required list<common.TConsensusGroupId> schemaRegionIdList
+ 2: required string tableName
+ 3: required binary patternInfo
+}
+
+struct TTableDeviceInvalidateCacheReq {
+ 1: required string database
+ 2: required string tableName
+ 3: required binary patternInfo
+}
+
struct TTsFilePieceReq {
1: required binary body
2: required string uuid
@@ -1082,6 +1101,31 @@ service IDataNodeRPCService {
*/
common.TSStatus deleteDevicesForDropTable(TDeleteDataOrDevicesForDropTableReq req)
+ /**
+ * Construct table device black list
+ */
+ common.TSStatus constructTableDeviceBlackList(TTableDeviceDeletionWithPatternAndFilterReq req)
+
+ /**
+ * Rollback table device black list
+ */
+ common.TSStatus rollbackTableDeviceBlackList(TTableDeviceDeletionWithPatternReq req)
+
+ /**
+ * Delete data for table devices
+ */
+ common.TSStatus invalidateMatchedTableDeviceCache(TTableDeviceInvalidateCacheReq req)
+
+ /**
+ * Delete data for table devices
+ */
+ common.TSStatus deleteDataForTableDevice(TTableDeviceDeletionWithPatternAndFilterReq req)
+
+ /**
+ * Delete table devices in black list
+ */
+ common.TSStatus deleteTableDeviceInBlackList(TTableDeviceDeletionWithPatternReq req)
+
common.TTestConnectionResp submitTestConnectionTask(common.TNodeLocations nodeLocations)
/** Empty rpc, only for connection test */