OneSizeFitsQuorum commented on code in PR #14736:
URL: https://github.com/apache/iotdb/pull/14736#discussion_r1923025051
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java:
##########
@@ -155,132 +146,18 @@ public void loadSnapshot(File latestSnapshotRootDir) {
}
protected PlanNode grabPlanNode(IndexedConsensusRequest indexedRequest) {
- List<InsertNode> insertNodes = new
ArrayList<>(indexedRequest.getRequests().size());
- List<DeleteDataNode> deleteDataNodes = new ArrayList<>();
+ List<SearchNode> searchNodes = new ArrayList<>();
for (IConsensusRequest req : indexedRequest.getRequests()) {
// PlanNode in IndexedConsensusRequest should always be InsertNode
PlanNode planNode = getPlanNode(req);
if (planNode instanceof SearchNode) {
((SearchNode)
planNode).setSearchIndex(indexedRequest.getSearchIndex());
- }
- if (planNode instanceof InsertNode) {
- insertNodes.add((InsertNode) planNode);
- } else if (planNode instanceof DeleteDataNode) {
- deleteDataNodes.add((DeleteDataNode) planNode);
- } else if (indexedRequest.getRequests().size() == 1) {
- // If the planNode is not InsertNode, it is expected that the
IndexedConsensusRequest only
- // contains one request
- return planNode;
+ searchNodes.add((SearchNode) planNode);
} else {
- throw new IllegalArgumentException(
- "PlanNodes in IndexedConsensusRequest are not InsertNode and "
- + "the size of requests are larger than 1");
- }
- }
- if (!insertNodes.isEmpty()) {
- if (!deleteDataNodes.isEmpty()) {
- throw new IllegalArgumentException(
- "One indexedRequest cannot contain InsertNode and DeleteDataNode
at the same time");
+ logger.warn("Unexpected plan node type {}", planNode.getClass());
}
- return mergeInsertNodes(insertNodes);
- }
- return mergeDeleteDataNode(deleteDataNodes);
- }
-
- private DeleteDataNode mergeDeleteDataNode(List<DeleteDataNode>
deleteDataNodes) {
- int size = deleteDataNodes.size();
- if (size == 0) {
- throw new IllegalArgumentException("deleteDataNodes is empty");
- }
- DeleteDataNode firstOne = deleteDataNodes.get(0);
- if (size == 1) {
- return firstOne;
- }
- if (!deleteDataNodes.stream()
- .allMatch(
- deleteDataNode ->
- firstOne.getDeleteStartTime() ==
deleteDataNode.getDeleteStartTime()
- && firstOne.getDeleteEndTime() ==
deleteDataNode.getDeleteEndTime())) {
- throw new IllegalArgumentException(
- "DeleteDataNodes which start time or end time are not same cannot be
merged");
- }
- List<MeasurementPath> pathList =
- deleteDataNodes.stream()
- .flatMap(deleteDataNode -> deleteDataNode.getPathList().stream())
- // Some time the deleteDataNode list contains a path for multiple
times, so use
- // distinct() to clear them
- .distinct()
- .collect(Collectors.toList());
- return new DeleteDataNode(
- firstOne.getPlanNodeId(),
- pathList,
- firstOne.getDeleteStartTime(),
- firstOne.getDeleteEndTime());
- }
-
- /**
- * Merge insert nodes sharing same search index ( e.g. tablet-100,
tablet-100, tablet-100 will be
- * merged to one multi-tablet). <br>
- * Notice: the continuity of insert nodes sharing same search index should
be protected by the
- * upper layer.
- *
- * @exception IllegalArgumentException when insertNodes is empty
- */
- protected InsertNode mergeInsertNodes(List<InsertNode> insertNodes) {
- int size = insertNodes.size();
- if (size == 0) {
- throw new IllegalArgumentException("insertNodes should never be empty");
}
- if (size == 1) {
- return insertNodes.get(0);
- }
-
- InsertNode result;
- List<Integer> index = new ArrayList<>();
- int i = 0;
- switch (insertNodes.get(0).getType()) {
- case RELATIONAL_INSERT_TABLET:
- case INSERT_TABLET:
- // merge to InsertMultiTabletsNode
- List<InsertTabletNode> insertTabletNodes = new ArrayList<>(size);
- for (InsertNode insertNode : insertNodes) {
- insertTabletNodes.add((InsertTabletNode) insertNode);
- index.add(i);
- i++;
- }
- result =
- new InsertMultiTabletsNode(
- insertNodes.get(0).getPlanNodeId(), index, insertTabletNodes);
- break;
- case INSERT_ROW:
- // merge to InsertRowsNode
- List<InsertRowNode> insertRowNodes = new ArrayList<>(size);
- for (InsertNode insertNode : insertNodes) {
- insertRowNodes.add((InsertRowNode) insertNode);
- index.add(i);
- i++;
- }
- result = new InsertRowsNode(insertNodes.get(0).getPlanNodeId(), index,
insertRowNodes);
- break;
- case INSERT_ROWS:
- // merge to InsertRowsNode
- List<InsertRowNode> list = new ArrayList<>();
- for (InsertNode insertNode : insertNodes) {
- for (InsertRowNode insertRowNode : ((InsertRowsNode)
insertNode).getInsertRowNodeList()) {
- list.add(insertRowNode);
- index.add(i);
- i++;
- }
- }
- result = new InsertRowsNode(insertNodes.get(0).getPlanNodeId(), index,
list);
- break;
- default:
- throw new UnSupportedDataTypeException(
- "Unsupported node type " + insertNodes.get(0).getType());
- }
- result.setSearchIndex(insertNodes.get(0).getSearchIndex());
- result.setTargetPath(insertNodes.get(0).getTargetPath());
- return result;
+ return searchNodes.get(0).merge(searchNodes);
Review Comment:
Add some validation in the function entry to ensure that no NPE errors are
reported?
--
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]
