JackieTien97 commented on code in PR #18127:
URL: https://github.com/apache/iotdb/pull/18127#discussion_r3534093133
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RemoveDataNodeHandler.java:
##########
@@ -455,6 +456,9 @@ public void
removeDataNodePersistence(List<TDataNodeLocation> removedDataNodes)
PartitionMetrics.unbindDataNodePartitionMetricsWhenUpdate(
MetricService.getInstance(),
NodeUrlUtils.convertTEndPointUrl(dataNodeLocation.getClientRpcEndPoint()));
+ // Drop the removed DataNode's metadata-lease contact/capability state
so it is not retained,
+ // and a future DataNode reusing the id cannot inherit stale fencing
history.
Review Comment:
id will always increase, no new DN will reuse removed DN's id
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java:
##########
@@ -2416,12 +2416,17 @@ public TDeleteTableDeviceResp deleteDevices(
}
}
+ // only care about the AbstractAlterOrDropTableProcedure(except the drop
table/view)
+ // and the DeleteDatabaseProcedure
Review Comment:
explain why we need to exclude the drop table/view procedure
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractAlterOrDropTableProcedure.java:
##########
@@ -91,17 +93,22 @@ protected void preRelease(final ConfigNodeProcedureEnv env)
{
}
protected void preRelease(final ConfigNodeProcedureEnv env, final @Nullable
String oldName) {
- final Map<Integer, TSStatus> failedResults =
- SchemaUtils.preReleaseTable(database, table, env.getConfigManager(),
oldName);
-
- if (!failedResults.isEmpty()) {
- // All dataNodes must clear the related schema cache
+ // Proceed once every unreachable DataNode is provably self-fenced instead
of hard-failing the
+ // DDL: a fenced DataNode fails closed on its now-stale table cache and
resyncs on lease
+ // recovery, so it cannot serve dirty schema. Only fail if an unacked
DataNode is not provably
+ // fenced (it may still be serving clients).
+ final TUpdateTableReq req = SchemaUtils.BuildPreUpdateTableReq(database,
table, oldName);
Review Comment:
```suggestion
final TUpdateTableReq req = SchemaUtils.buildPreUpdateTableReq(database,
table, oldName);
```
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AbstractAlterOrDropTableProcedure.java:
##########
@@ -140,18 +147,21 @@ protected void rollbackPreRelease(final
ConfigNodeProcedureEnv env) {
protected void rollbackPreRelease(
final ConfigNodeProcedureEnv env, final @Nullable String tableName) {
- final Map<Integer, TSStatus> failedResults =
- SchemaUtils.rollbackPreRelease(
- database, table.getTableName(), env.getConfigManager(), tableName);
-
- if (!failedResults.isEmpty()) {
- // All dataNodes must clear the related schema cache
+ // A down DataNode must not block rollback either: proceed past
provably-fenced DataNodes (which
+ // resync on recovery) and only fail on an unacked DataNode that is not
provably fenced.
+ final TUpdateTableReq req =
+ SchemaUtils.rollbackUpdateTableReq(database, table.getTableName(),
tableName);
+ final boolean proceeded =
+ new ClusterCachePropagator(env.getConfigManager())
+ .propagate(targets -> SchemaUtils.broadcastTableUpdate(req,
targets));
+
+ if (!proceeded) {
LOGGER.warn(
ProcedureMessages.FAILED_TO_ROLLBACK_PRE_RELEASE_FOR_TABLE_INFO_TO_DATANODE,
getActionMessage(),
database,
table.getTableName(),
- failedResults);
+ "an unreachable DataNode is not provably fenced");
Review Comment:
follow the i18n standarn, define a constant in both zh and en.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java:
##########
@@ -151,16 +153,22 @@ protected void preCreateTable(final
ConfigNodeProcedureEnv env) {
}
private void preReleaseTable(final ConfigNodeProcedureEnv env) {
- final Map<Integer, TSStatus> failedResults =
- SchemaUtils.preReleaseTable(database, table, env.getConfigManager(),
null);
-
- if (!failedResults.isEmpty()) {
- // All dataNodes must clear the related schema cache
+ // Broadcast the pre-update to all DataNodes. Instead of failing whenever
any DataNode is
+ // unreachable, proceed once every unacked DataNode is provably
self-fenced: such a DataNode
+ // fails closed on its (now-stale) table cache and resyncs on lease
recovery, so it cannot serve
+ // dirty schema. Only fail if an unacked DataNode is not provably fenced
(it may still be
+ // serving clients).
+ final TUpdateTableReq req = SchemaUtils.BuildPreUpdateTableReq(database,
table, null);
+ final boolean proceeded =
+ new ClusterCachePropagator(env.getConfigManager())
+ .propagate(targets -> SchemaUtils.broadcastTableUpdate(req,
targets));
+
+ if (!proceeded) {
LOGGER.warn(
ProcedureMessages.FAILED_TO_SYNC_TABLE_PRE_CREATE_INFO_TO_DATANODE_FAILURE,
database,
table.getTableName(),
- failedResults);
+ "an unreachable DataNode is not provably fenced");
Review Comment:
i18n
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/AlterTimeSeriesDataTypeProcedure.java:
##########
@@ -249,26 +246,17 @@ public static void invalidateCache(
final String requestMessage,
final Consumer<ProcedureException> setFailure,
final boolean needLock) {
- final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
-
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
- final DataNodeAsyncRequestContext<TInvalidateMatchedSchemaCacheReq,
TSStatus> clientHandler =
- new DataNodeAsyncRequestContext<>(
- CnToDnAsyncRequestType.INVALIDATE_MATCHED_SCHEMA_CACHE,
- new
TInvalidateMatchedSchemaCacheReq(measurementPathBytes).setNeedLock(needLock),
- dataNodeLocationMap);
-
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
- final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
- for (final TSStatus status : statusMap.values()) {
+ // Proceed past provably-fenced DataNodes instead of hard-failing on the
first unreachable one
+ // (see SchemaUtils.invalidateMatchedSchemaCache). Runs before the
physical datatype change, so
+ // the "alter only after PROCEED" ordering holds.
+ if (!SchemaUtils.invalidateMatchedSchemaCache(
+ env.getConfigManager(), measurementPathBytes, needLock)) {
// All dataNodes must clear the related schemaEngine cache
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.error(
-
ProcedureMessages.FAILED_TO_INVALIDATE_SCHEMAENGINE_CACHE_OF_TIMESERIES,
- requestMessage);
- setFailure.accept(
- new ProcedureException(
- new
MetadataException(ProcedureMessages.INVALIDATE_SCHEMAENGINE_CACHE_FAILED)));
- return;
- }
+ LOGGER.error(
Review Comment:
warn
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteLogicalViewProcedure.java:
##########
@@ -167,26 +164,15 @@ protected List<TConsensusGroupId>
processResponseOfOneDataNode(
}
private void invalidateCache(final ConfigNodeProcedureEnv env) {
- final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
-
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
- final DataNodeAsyncRequestContext<TInvalidateMatchedSchemaCacheReq,
TSStatus> clientHandler =
- new DataNodeAsyncRequestContext<>(
- CnToDnAsyncRequestType.INVALIDATE_MATCHED_SCHEMA_CACHE,
- new TInvalidateMatchedSchemaCacheReq(patternTreeBytes),
- dataNodeLocationMap);
-
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
- final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
- for (final TSStatus status : statusMap.values()) {
+ if (!SchemaUtils.invalidateMatchedSchemaCache(
+ env.getConfigManager(), patternTreeBytes, false)) {
// all dataNodes must clear the related schemaengine cache
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.error(
- ProcedureMessages.FAILED_TO_INVALIDATE_SCHEMAENGINE_CACHE_OF_VIEW,
requestMessage);
- setFailure(
- new ProcedureException(
- new MetadataException(
-
ProcedureMessages.INVALIDATE_VIEW_SCHEMAENGINE_CACHE_FAILED)));
- return;
- }
+ LOGGER.error(
Review Comment:
warn
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableColumnProcedure.java:
##########
@@ -150,43 +151,48 @@ private void checkAndPreDeleteColumn(final
ConfigNodeProcedureEnv env) {
}
private void invalidateCache(final ConfigNodeProcedureEnv env) {
- final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
-
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
- final DataNodeAsyncRequestContext<TInvalidateColumnCacheReq, TSStatus>
clientHandler =
- new DataNodeAsyncRequestContext<>(
- CnToDnAsyncRequestType.INVALIDATE_COLUMN_CACHE,
- new TInvalidateColumnCacheReq(database, tableName, columnName,
isAttributeColumn),
- 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(
- ProcedureMessages.FAILED_TO_INVALIDATE_COLUMN_S_CACHE_OF_TABLE,
- isAttributeColumn ? "attribute" : "measurement",
- columnName,
- database,
- tableName);
- setFailure(
- new ProcedureException(
- new MetadataException(
- String.format(
-
ProcedureMessages.INVALIDATE_COLUMN_CACHE_FAILED_FOR_TABLE,
- columnName,
- database,
- tableName))));
- return;
- }
- }
+ TInvalidateColumnCacheReq req =
+ new TInvalidateColumnCacheReq(database, tableName, columnName,
isAttributeColumn);
+ final boolean proceeded =
+ new ClusterCachePropagator(env.getConfigManager())
+ .propagate(targets -> broadCastInvalidateCache(req, targets));
+ if (!proceeded) {
+ LOGGER.error(
Review Comment:
no need to be error, just warn is enough, error only for unexpected fatal
cases.
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/PreDeleteTsTable.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.schema.table;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+public class PreDeleteTsTable extends TsTable {
+ public static final int PRE_DELETE_MARKER = -2;
Review Comment:
```suggestion
public static final int PRE_DELETE_MARKER = -3;
```
put it as an enum, add comments that -2 is reserved and don't use.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/AlterLogicalViewProcedure.java:
##########
@@ -123,27 +120,15 @@ protected Flow executeFromState(
}
private void invalidateCache(final ConfigNodeProcedureEnv env) {
- final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
-
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
- final DataNodeAsyncRequestContext<TInvalidateMatchedSchemaCacheReq,
TSStatus> clientHandler =
- new DataNodeAsyncRequestContext<>(
- CnToDnAsyncRequestType.INVALIDATE_MATCHED_SCHEMA_CACHE,
- new TInvalidateMatchedSchemaCacheReq(patternTreeBytes),
- dataNodeLocationMap);
-
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
- final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
- for (final TSStatus status : statusMap.values()) {
+ if (!SchemaUtils.invalidateMatchedSchemaCache(
+ env.getConfigManager(), patternTreeBytes, false)) {
// all dataNodes must clear the related schemaengine cache
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.error(
- ProcedureMessages.FAILED_TO_INVALIDATE_SCHEMAENGINE_CACHE_OF_VIEW,
- viewPathToSourceMap.keySet());
- setFailure(
- new ProcedureException(
- new MetadataException(
-
ProcedureMessages.INVALIDATE_VIEW_SCHEMAENGINE_CACHE_FAILED)));
- return;
- }
+ LOGGER.error(
Review Comment:
warn?
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteTimeSeriesProcedure.java:
##########
@@ -197,26 +194,18 @@ public static void invalidateCache(
final String requestMessage,
final Consumer<ProcedureException> setFailure,
final boolean needLock) {
- final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
-
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
- final DataNodeAsyncRequestContext<TInvalidateMatchedSchemaCacheReq,
TSStatus> clientHandler =
- new DataNodeAsyncRequestContext<>(
- CnToDnAsyncRequestType.INVALIDATE_MATCHED_SCHEMA_CACHE,
- new
TInvalidateMatchedSchemaCacheReq(patternTreeBytes).setNeedLock(needLock),
- dataNodeLocationMap);
-
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
- final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
- for (final TSStatus status : statusMap.values()) {
+ // Proceed once every unreachable DataNode is provably self-fenced (it
fails closed on its
+ // schema cache and resyncs on recovery, so it cannot serve the
to-be-deleted/altered series),
+ // instead of hard-failing on the first unreachable DataNode. This runs
before the physical
+ // delete in the state machine, so the "delete only after PROCEED"
ordering holds.
+ if (!SchemaUtils.invalidateMatchedSchemaCache(
+ env.getConfigManager(), patternTreeBytes, needLock)) {
// All dataNodes must clear the related schemaEngine cache
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.error(
-
ProcedureMessages.FAILED_TO_INVALIDATE_SCHEMAENGINE_CACHE_OF_TIMESERIES,
- requestMessage);
- setFailure.accept(
- new ProcedureException(
- new
MetadataException(ProcedureMessages.INVALIDATE_SCHEMAENGINE_CACHE_FAILED)));
- return;
- }
+ LOGGER.error(
Review Comment:
warn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
