Skip to content

Commit d2b97bc

Browse files
authored
Reuse DataRegion insert lock for IoTConsensus shutdown fencing (#18893)
1 parent a7b88b1 commit d2b97bc

7 files changed

Lines changed: 422 additions & 83 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,6 @@ public TSStatus visitRelationalInsertRows(RelationalInsertRowsNode node, DataReg
7878
public TSStatus visitInsertRow(InsertRowNode node, DataRegion dataRegion) {
7979
try {
8080
dataRegion.insert(node);
81-
dataRegion.insertSeparatorToWAL(node);
8281
return StatusUtils.OK;
8382
} catch (OutOfTTLException e) {
8483
LOGGER.warn(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE_CAUSED, node, e.getMessage());
@@ -104,7 +103,6 @@ public TSStatus visitRelationalInsertTablet(
104103
public TSStatus visitInsertTablet(final InsertTabletNode node, final DataRegion dataRegion) {
105104
try {
106105
dataRegion.insertTablet(node);
107-
dataRegion.insertSeparatorToWAL(node);
108106
return StatusUtils.OK;
109107
} catch (final OutOfTTLException e) {
110108
LOGGER.debug(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE_CAUSED, node, e.getMessage());
@@ -144,7 +142,6 @@ public TSStatus visitInsertTablet(final InsertTabletNode node, final DataRegion
144142
public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) {
145143
try {
146144
dataRegion.insert(node);
147-
dataRegion.insertSeparatorToWAL(node);
148145
return StatusUtils.OK;
149146
} catch (WriteProcessRejectException e) {
150147
LOGGER.warn(DataNodeMiscMessages.REJECT_EXECUTING_PLAN_NODE, node, e.getMessage());
@@ -184,7 +181,6 @@ public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) {
184181
public TSStatus visitInsertMultiTablets(InsertMultiTabletsNode node, DataRegion dataRegion) {
185182
try {
186183
dataRegion.insertTablets(node);
187-
dataRegion.insertSeparatorToWAL(node);
188184
return StatusUtils.OK;
189185
} catch (WriteProcessRejectException e) {
190186
LOGGER.warn(DataNodeMiscMessages.REJECT_EXECUTING_PLAN_NODE, node, e.getMessage());
@@ -222,7 +218,6 @@ public TSStatus visitInsertRowsOfOneDevice(
222218
InsertRowsOfOneDeviceNode node, DataRegion dataRegion) {
223219
try {
224220
dataRegion.insert(node);
225-
dataRegion.insertSeparatorToWAL(node);
226221
return StatusUtils.OK;
227222
} catch (WriteProcessRejectException e) {
228223
LOGGER.warn(DataNodeMiscMessages.REJECT_EXECUTING_PLAN_NODE, node, e.getMessage());
@@ -284,6 +279,8 @@ public TSStatus visitDeleteData(DeleteDataNode node, DataRegion dataRegion) {
284279
}
285280
dataRegion.insertSeparatorToWAL(node);
286281
return StatusUtils.OK;
282+
} catch (WriteProcessRejectException e) {
283+
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
287284
} catch (IOException | IllegalPathException e) {
288285
LOGGER.error(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE, node, e);
289286
return new TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode());
@@ -297,8 +294,9 @@ public TSStatus visitDeleteData(
297294
final RelationalDeleteDataNode node, final DataRegion dataRegion) {
298295
try {
299296
dataRegion.deleteByTable(node);
300-
dataRegion.insertSeparatorToWAL(node);
301297
return StatusUtils.OK;
298+
} catch (final WriteProcessRejectException e) {
299+
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
302300
} catch (final IOException e) {
303301
LOGGER.error(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE, node, e);
304302
return new TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode());

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -251,7 +251,7 @@ protected TSStatus write(PlanNode planNode) {
251251
TSStatus result = null;
252252
int retryTime = 0;
253253
while (retryTime < MAX_WRITE_RETRY_TIMES) {
254-
result = executeWrite(planNode);
254+
result = planNode.accept(new DataExecutionVisitor(), region);
255255
// Let pipe retry with the original event instead of retrying a possibly mutated node here.
256256
if (needRetryForSpecificCases(result.getCode(), planNode)) {
257257
retryTime++;
@@ -284,10 +284,6 @@ protected TSStatus write(PlanNode planNode) {
284284
return result;
285285
}
286286

287-
protected TSStatus executeWrite(PlanNode planNode) {
288-
return planNode.accept(new DataExecutionVisitor(), region);
289-
}
290-
291287
protected void waitBeforeNextWriteRetry() throws InterruptedException {
292288
Thread.sleep(WRITE_RETRY_WAIT_TIME_IN_MS);
293289
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusDataRegionStateMachine.java‎

Lines changed: 0 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -50,13 +50,6 @@ public IoTConsensusDataRegionStateMachine(DataRegion region) {
5050
super(region);
5151
}
5252

53-
@Override
54-
protected TSStatus executeWrite(PlanNode planNode) {
55-
final DataRegion targetRegion = region;
56-
return targetRegion.executeIoTConsensusWrite(
57-
() -> planNode.accept(new DataExecutionVisitor(), targetRegion));
58-
}
59-
6053
@Override
6154
public TSStatus write(IConsensusRequest request) {
6255
try {

0 commit comments

Comments
 (0)