Skip to content

Optimized the error log for schema execution #16982

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Caideyipi merged 10 commits into from
Jan 6, 2026
Merged

Optimized the error log for schema execution #16982

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
5b9f189
userE
Caideyipi Jan 5, 2026
26d673e
logger
Caideyipi Jan 5, 2026
1ce0f32
fix
Caideyipi Jan 5, 2026
283db33
Merge branch 'master' of https://github.com/apache/iotdb into err-cli…
Caideyipi Jan 6, 2026
d36f584
refactor
Caideyipi Jan 6, 2026
95eae5e
tsd
Caideyipi Jan 6, 2026
85d42df
ut
Caideyipi Jan 6, 2026
2f3789c
fix
Caideyipi Jan 6, 2026
e9c3ca7
logic-fix
Caideyipi Jan 6, 2026
adc0931
cut
Caideyipi Jan 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,18 @@ public static TSStatus getStatus(List<TSStatus> statusList) {
return status;
}

public static TSStatus extractFailureStatues(final TSStatus input) {
return input.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()
? new TSStatus(input.getCode())
.setMessage(input.getMessage())
.setSubStatus(
input.getSubStatus().stream()
.filter(
status -> status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode())
.collect(Collectors.toList()))
: input;
}

/**
* Convert from {@link TSStatusCode} to {@link TSStatus}, which has message appended with existing
* status message
Expand Down
8 changes: 3 additions & 5 deletions ...a/org/apache/iotdb/confignode/procedure/impl/schema/AlterEncodingCompressorProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
import org.apache.iotdb.mpp.rpc.thrift.TAlterEncodingCompressorReq;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;

import org.apache.tsfile.utils.ReadWriteIOUtils;
Expand All @@ -51,7 +52,6 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
Expand Down Expand Up @@ -176,8 +176,6 @@ private boolean alterEncodingCompressorInSchemaRegion(final ConfigNodeProcedureE
.setCompressor(compressor)
.setEncoding(encoding))) {

private final Map<TDataNodeLocation, TSStatus> failureMap = new HashMap<>();

@Override
protected List<TConsensusGroupId> processResponseOfOneDataNode(
final TDataNodeLocation dataNodeLocation,
Expand All @@ -203,7 +201,7 @@ protected List<TConsensusGroupId> processResponseOfOneDataNode(
failedRegionList.addAll(consensusGroupIdList);
}
if (!failedRegionList.isEmpty()) {
failureMap.put(dataNodeLocation, response);
failureMap.put(dataNodeLocation, RpcUtils.extractFailureStatues(response));
} else {
failureMap.remove(dataNodeLocation);
}
Expand All @@ -219,7 +217,7 @@ protected void onAllReplicasetFailure(
new MetadataException(
String.format(
"Alter encoding compressor %s in schema regions failed. Failures: %s",
requestMessage, failureMap))));
requestMessage, printFailureMap()))));
interruptTask();
}
};
Expand Down
10 changes: 8 additions & 2 deletions ...ain/java/org/apache/iotdb/confignode/procedure/impl/schema/AlterLogicalViewProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@
import java.util.Objects;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.stream.Collectors;

public class AlterLogicalViewProcedure
extends StateMachineProcedure<ConfigNodeProcedureEnv, AlterLogicalViewState> {
Expand All @@ -72,6 +73,8 @@ public class AlterLogicalViewProcedure
private transient PathPatternTree pathPatternTree;
private transient ByteBuffer patternTreeBytes;

protected final Map<TDataNodeLocation, TSStatus> failureMap = new HashMap<>();

public AlterLogicalViewProcedure(final boolean isGeneratedByPipe) {
super(isGeneratedByPipe);
}
Expand Down Expand Up @@ -390,11 +393,14 @@ protected void onAllReplicasetFailure(
new ProcedureException(
new MetadataException(
String.format(
"Alter view %s failed when [%s] because failed to execute in all replicaset of schemaRegion %s. Failure nodes: %s",
"Alter view %s failed when [%s] because failed to execute in all replicaset of schemaRegion %s. Failure nodes: %s, statuses: %s",
viewPathToSourceMap.keySet(),
taskName,
consensusGroupId.id,
dataNodeLocationSet))));
dataNodeLocationSet.stream()
.map(TDataNodeLocation::getDataNodeId)
.collect(Collectors.toSet()),
failureStatusList))));
interruptTask();
}
}
Expand Down
8 changes: 3 additions & 5 deletions ...a/org/apache/iotdb/confignode/procedure/impl/schema/AlterTimeSeriesDataTypeProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import org.apache.iotdb.mpp.rpc.thrift.TAlterTimeSeriesReq;
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateMatchedSchemaCacheReq;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;

import org.apache.tsfile.enums.TSDataType;
Expand All @@ -54,7 +55,6 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
Expand Down Expand Up @@ -195,8 +195,6 @@ private boolean alterTimeSeriesDataType(final ConfigNodeProcedureEnv env) {
ByteBuffer.wrap(stream.toByteArray()));
})) {

private final Map<TDataNodeLocation, TSStatus> failureMap = new HashMap<>();

@Override
protected List<TConsensusGroupId> processResponseOfOneDataNode(
final TDataNodeLocation dataNodeLocation,
Expand All @@ -221,7 +219,7 @@ protected List<TConsensusGroupId> processResponseOfOneDataNode(
failedRegionList.addAll(consensusGroupIdList);
}
if (!failedRegionList.isEmpty()) {
failureMap.put(dataNodeLocation, response);
failureMap.put(dataNodeLocation, RpcUtils.extractFailureStatues(response));
} else {
failureMap.remove(dataNodeLocation);
}
Expand All @@ -240,7 +238,7 @@ protected void onAllReplicasetFailure(
measurementPath.getFullPath(),
measurementPath.getSeriesType(),
dataType,
failureMap))));
printFailureMap()))));
interruptTask();
}
};
Expand Down
17 changes: 16 additions & 1 deletion ...in/java/org/apache/iotdb/confignode/procedure/impl/schema/DataNodeRegionTaskExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,11 +22,13 @@
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.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.manager.ConfigManager;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.rpc.TSStatusCode;

import java.util.ArrayList;
import java.util.HashMap;
Expand All @@ -35,6 +37,7 @@
import java.util.Map;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.stream.Collectors;

import static org.apache.iotdb.confignode.procedure.impl.schema.DataNodeRegionGroupUtil.getAllReplicaDataNodeRegionGroupMap;
import static org.apache.iotdb.confignode.procedure.impl.schema.DataNodeRegionGroupUtil.getLeaderDataNodeRegionGroupMap;
Expand All @@ -48,7 +51,7 @@ public abstract class DataNodeRegionTaskExecutor<Q, R> {
protected final CnToDnAsyncRequestType dataNodeRequestType;
protected final BiFunction<TDataNodeLocation, List<TConsensusGroupId>, Q>
dataNodeRequestGenerator;

protected final Map<TDataNodeLocation, TSStatus> failureMap = new HashMap<>();
private boolean isInterrupted = false;

protected DataNodeRegionTaskExecutor(
Expand Down Expand Up @@ -229,4 +232,16 @@ protected abstract List<TConsensusGroupId> processResponseOfOneDataNode(
*/
protected abstract void onAllReplicasetFailure(
final TConsensusGroupId consensusGroupId, final Set<TDataNodeLocation> dataNodeLocationSet);

protected String printFailureMap() {
return failureMap.entrySet().stream()
.collect(
Collectors.toMap(
entry -> "DataNodeId: " + entry.getKey().getDataNodeId(),
entry ->
entry.getValue().getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()
? entry.getValue().getSubStatus()
: entry.getValue()))
.toString();
}
}
39 changes: 39 additions & 0 deletions .../java/org/apache/iotdb/confignode/procedure/impl/schema/DataNodeTSStatusTaskExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;

import java.util.ArrayList;
Expand All @@ -34,6 +35,7 @@

public abstract class DataNodeTSStatusTaskExecutor<Q>
extends DataNodeRegionTaskExecutor<Q, TSStatus> {
protected List<TSStatus> successResult = new ArrayList<>();

protected DataNodeTSStatusTaskExecutor(
final ConfigNodeProcedureEnv env,
Expand All @@ -56,6 +58,7 @@ protected List<TConsensusGroupId> processResponseOfOneDataNode(
final TSStatus response) {
final List<TConsensusGroupId> failedRegionList = new ArrayList<>();
if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
failureMap.remove(dataNodeLocation);
return failedRegionList;
}

Expand All @@ -69,6 +72,42 @@ protected List<TConsensusGroupId> processResponseOfOneDataNode(
} else {
failedRegionList.addAll(consensusGroupIdList);
}
if (!failedRegionList.isEmpty()) {
failureMap.put(dataNodeLocation, RpcUtils.extractFailureStatues(response));
} else {
failureMap.remove(dataNodeLocation);
}
return failedRegionList;
}

protected List<TConsensusGroupId> processResponseOfOneDataNodeWithSuccessResult(
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()) {
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);
}
if (!failedRegionList.isEmpty()) {
failureMap.put(dataNodeLocation, RpcUtils.extractFailureStatues(response));
} else {
failureMap.remove(dataNodeLocation);
}
return failedRegionList;
}

public List<TSStatus> getSuccessResult() {
return successResult;
}
}
32 changes: 8 additions & 24 deletions ...n/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,6 @@ private long constructBlackList(ConfigNodeProcedureEnv env) {
if (targetSchemaRegionGroup.isEmpty()) {
return 0;
}
List<TSStatus> successResult = new ArrayList<>();
DeactivateTemplateRegionTaskExecutor<TConstructSchemaBlackListWithTemplateReq>
constructBlackListTask =
new DeactivateTemplateRegionTaskExecutor<TConstructSchemaBlackListWithTemplateReq>(
Expand All @@ -156,26 +155,11 @@ private long constructBlackList(ConfigNodeProcedureEnv env) {
consensusGroupIdList, dataNodeRequest))) {
@Override
protected List<TConsensusGroupId> processResponseOfOneDataNode(
TDataNodeLocation dataNodeLocation,
List<TConsensusGroupId> consensusGroupIdList,
TSStatus response) {
List<TConsensusGroupId> failedRegionList = new ArrayList<>();
if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
successResult.add(response);
} else if (response.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
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;
final TDataNodeLocation dataNodeLocation,
final List<TConsensusGroupId> consensusGroupIdList,
final TSStatus response) {
return processResponseOfOneDataNodeWithSuccessResult(
dataNodeLocation, consensusGroupIdList, response);
}
};
constructBlackListTask.execute();
Expand All @@ -185,7 +169,7 @@ protected List<TConsensusGroupId> processResponseOfOneDataNode(
}

long preDeletedNum = 0;
for (TSStatus resp : successResult) {
for (TSStatus resp : constructBlackListTask.getSuccessResult()) {
preDeletedNum += Long.parseLong(resp.getMessage());
}
return preDeletedNum;
Expand Down Expand Up @@ -465,12 +449,12 @@ protected void onAllReplicasetFailure(
new ProcedureException(
new MetadataException(
String.format(
"Deactivate template of %s failed when [%s] because failed to execute in all replicaset of %s %s. Failure nodes: %s",
"Deactivate template of %s failed when [%s] because failed to execute in all replicaset of %s %s. Failure: %s",
requestMessage,
taskName,
consensusGroupId.type,
consensusGroupId.id,
dataNodeLocationSet))));
printFailureMap()))));
interruptTask();
}
}
Expand Down
32 changes: 8 additions & 24 deletions ...in/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteLogicalViewProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,6 @@
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;
Expand Down Expand Up @@ -135,7 +134,6 @@ private long constructBlackList(ConfigNodeProcedureEnv env) {
if (targetSchemaRegionGroup.isEmpty()) {
return 0;
}
List<TSStatus> successResult = new ArrayList<>();
DeleteLogicalViewRegionTaskExecutor<TConstructViewSchemaBlackListReq> constructBlackListTask =
new DeleteLogicalViewRegionTaskExecutor<TConstructViewSchemaBlackListReq>(
"construct view schema engine black list",
Expand All @@ -146,25 +144,11 @@ private long constructBlackList(ConfigNodeProcedureEnv env) {
new TConstructViewSchemaBlackListReq(consensusGroupIdList, patternTreeBytes))) {
@Override
protected List<TConsensusGroupId> processResponseOfOneDataNode(
TDataNodeLocation dataNodeLocation,
List<TConsensusGroupId> consensusGroupIdList,
TSStatus response) {
List<TConsensusGroupId> failedRegionList = new ArrayList<>();
if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
successResult.add(response);
} else if (response.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
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;
final TDataNodeLocation dataNodeLocation,
final List<TConsensusGroupId> consensusGroupIdList,
final TSStatus response) {
return processResponseOfOneDataNodeWithSuccessResult(
dataNodeLocation, consensusGroupIdList, response);
}
};
constructBlackListTask.execute();
Expand All @@ -173,7 +157,7 @@ protected List<TConsensusGroupId> processResponseOfOneDataNode(
return 0;
}

return successResult.stream()
return constructBlackListTask.getSuccessResult().stream()
.mapToLong(resp -> Long.parseLong(resp.getMessage()))
.reduce(0L, Long::sum);
}
Expand Down Expand Up @@ -357,8 +341,8 @@ protected void onAllReplicasetFailure(
new ProcedureException(
new MetadataException(
String.format(
"Delete view %s failed when [%s] because failed to execute in all replicaset of schemaRegion %s. Failure nodes: %s",
requestMessage, taskName, consensusGroupId.id, dataNodeLocationSet))));
"Delete view %s failed when [%s] because failed to execute in all replicaset of schemaRegion %s. Failures: %s",
requestMessage, taskName, consensusGroupId.id, printFailureMap()))));
interruptTask();
}
}
Expand Down
Loading