diff --git a/.mvn/develocity.xml b/.mvn/develocity.xml index 1e82305ef93c5..b505d1a366647 100644 --- a/.mvn/develocity.xml +++ b/.mvn/develocity.xml @@ -26,9 +26,7 @@ #{isFalse(env['GITHUB_ACTIONS'])} - - - + #{{'0.0.0.0'}} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java index 35d82a5454ed7..b16314ff5ad3f 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java @@ -466,7 +466,7 @@ private TSStatus consensusWritePartitionResult(ConfigPhysicalPlan plan) { return getConsensusManager().write(plan); } catch (ConsensusException e) { // The allocation might fail due to consensus error - LOGGER.error("Write DataPartition allocation result failed because: {}", status); + LOGGER.error("Write partition allocation result failed because: {}", status); TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()); res.setMessage(e.getMessage()); return res; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/SeriesPartitionTable.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/SeriesPartitionTable.java index 890e46a451a6e..494da226e83cd 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/SeriesPartitionTable.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/SeriesPartitionTable.java @@ -33,27 +33,27 @@ import java.io.InputStream; import java.io.OutputStream; import java.nio.ByteBuffer; -import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Objects; -import java.util.TreeMap; import java.util.Vector; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentSkipListMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; public class SeriesPartitionTable { - private final TreeMap> seriesPartitionMap; + private final ConcurrentSkipListMap> + seriesPartitionMap; public SeriesPartitionTable() { - this.seriesPartitionMap = new TreeMap<>(); + this.seriesPartitionMap = new ConcurrentSkipListMap<>(); } public SeriesPartitionTable(Map> seriesPartitionMap) { - this.seriesPartitionMap = new TreeMap<>(seriesPartitionMap); + this.seriesPartitionMap = new ConcurrentSkipListMap<>(seriesPartitionMap); } public Map> getSeriesPartitionMap() { @@ -61,7 +61,7 @@ public Map> getSeriesPartitionMap() } public void putDataPartition(TTimePartitionSlot timePartitionSlot, TConsensusGroupId groupId) { - seriesPartitionMap.computeIfAbsent(timePartitionSlot, empty -> new ArrayList<>()).add(groupId); + seriesPartitionMap.computeIfAbsent(timePartitionSlot, empty -> new Vector<>()).add(groupId); } /** @@ -75,7 +75,6 @@ public boolean getDataPartition( TTimeSlotList partitionSlotList, SeriesPartitionTable seriesPartitionTable) { AtomicBoolean result = new AtomicBoolean(true); List partitionSlots = partitionSlotList.getTimePartitionSlots(); - if (partitionSlots.isEmpty()) { // Return all DataPartitions in one SeriesPartitionSlot // when the queried TimePartitionSlots are empty @@ -84,7 +83,8 @@ public boolean getDataPartition( boolean isNeedLeftAll = partitionSlotList.isNeedLeftAll(), isNeedRightAll = partitionSlotList.isNeedRightAll(); if (isNeedLeftAll || isNeedRightAll) { - // we need to calculate the leftMargin which contains all the time partition on the unclosed + // we need to calculate the leftMargin which contains all the time partition on the + // unclosed // left side: (-oo, leftMargin) // and the rightMargin which contains all the time partition on the unclosed right side: // (rightMargin, +oo) @@ -106,7 +106,6 @@ public boolean getDataPartition( }) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))); } - // Return the DataPartition for each match TimePartitionSlot partitionSlots.forEach( timePartitionSlot -> { @@ -119,7 +118,6 @@ public boolean getDataPartition( } }); } - return result.get(); } @@ -153,7 +151,7 @@ public TConsensusGroupId getPredecessorDataPartition(TTimePartitionSlot timePart * @return the timePartition's corresponding dataRegionIds. return the dataRegions which * timeslotIds are in the time range [startTimeSlotId, endTimeSlotId]. */ - List getRegionId( + public List getRegionId( TTimePartitionSlot startTimeSlotId, TTimePartitionSlot endTimeSlotId) { return seriesPartitionMap.entrySet().stream() .filter( @@ -164,7 +162,7 @@ List getRegionId( .collect(Collectors.toList()); } - List getTimeSlotList( + public List getTimeSlotList( TConsensusGroupId regionId, long startTime, long endTime) { if (regionId.getId() == -1) { return seriesPartitionMap.keySet().stream() @@ -214,14 +212,12 @@ public void createDataPartition( public synchronized List filterUnassignedDataPartitionSlots( List partitionSlots) { List result = new Vector<>(); - partitionSlots.forEach( timePartitionSlot -> { if (!seriesPartitionMap.containsKey(timePartitionSlot)) { result.add(timePartitionSlot); } }); - return result; } @@ -258,13 +254,11 @@ public void deserialize(ByteBuffer buffer) { for (int i = 0; i < timePartitionSlotNum; i++) { TTimePartitionSlot timePartitionSlot = ThriftCommonsSerDeUtils.deserializeTTimePartitionSlot(buffer); - int consensusGroupIdNum = buffer.getInt(); List consensusGroupIds = new Vector<>(); for (int j = 0; j < consensusGroupIdNum; j++) { consensusGroupIds.add(ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(buffer)); } - seriesPartitionMap.put(timePartitionSlot, consensusGroupIds); } } @@ -276,7 +270,6 @@ public void deserialize(InputStream inputStream, TProtocol protocol) for (int i = 0; i < timePartitionSlotNum; i++) { TTimePartitionSlot timePartitionSlot = new TTimePartitionSlot(); timePartitionSlot.read(protocol); - int consensusGroupIdNum = ReadWriteIOUtils.readInt(inputStream); List consensusGroupIds = new Vector<>(); for (int j = 0; j < consensusGroupIdNum; j++) { @@ -284,7 +277,6 @@ public void deserialize(InputStream inputStream, TProtocol protocol) consensusGroupId.read(protocol); consensusGroupIds.add(consensusGroupId); } - seriesPartitionMap.put(timePartitionSlot, consensusGroupIds); } }