From 92f1129fb4318fdb840298d549a2700d7e8dfb1f Mon Sep 17 00:00:00 2001 From: YongzaoDan <532741407@qq.com> Date: Thu, 6 Jun 2024 15:57:32 +0800 Subject: [PATCH 1/3] finish --- .../manager/partition/PartitionManager.java | 2 +- .../partition/SeriesPartitionTable.java | 330 +++++++++++------- 2 files changed, 203 insertions(+), 129 deletions(-) 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..b722f564b191b 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 @@ -42,26 +42,42 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; public class SeriesPartitionTable { + private final ReentrantReadWriteLock seriesPartitionMapLock; private final TreeMap> seriesPartitionMap; public SeriesPartitionTable() { + this.seriesPartitionMapLock = new ReentrantReadWriteLock(); this.seriesPartitionMap = new TreeMap<>(); } public SeriesPartitionTable(Map> seriesPartitionMap) { + this.seriesPartitionMapLock = new ReentrantReadWriteLock(); this.seriesPartitionMap = new TreeMap<>(seriesPartitionMap); } public Map> getSeriesPartitionMap() { - return seriesPartitionMap; + seriesPartitionMapLock.readLock().lock(); + try { + return seriesPartitionMap; + } finally { + seriesPartitionMapLock.readLock().unlock(); + } } public void putDataPartition(TTimePartitionSlot timePartitionSlot, TConsensusGroupId groupId) { - seriesPartitionMap.computeIfAbsent(timePartitionSlot, empty -> new ArrayList<>()).add(groupId); + seriesPartitionMapLock.writeLock().lock(); + try { + seriesPartitionMap + .computeIfAbsent(timePartitionSlot, empty -> new ArrayList<>()) + .add(groupId); + } finally { + seriesPartitionMapLock.writeLock().unlock(); + } } /** @@ -74,52 +90,56 @@ public void putDataPartition(TTimePartitionSlot timePartitionSlot, TConsensusGro public boolean getDataPartition( TTimeSlotList partitionSlotList, SeriesPartitionTable seriesPartitionTable) { AtomicBoolean result = new AtomicBoolean(true); + seriesPartitionMapLock.readLock().lock(); List partitionSlots = partitionSlotList.getTimePartitionSlots(); - - if (partitionSlots.isEmpty()) { - // Return all DataPartitions in one SeriesPartitionSlot - // when the queried TimePartitionSlots are empty - seriesPartitionTable.getSeriesPartitionMap().putAll(seriesPartitionMap); - } else { - 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 - // left side: (-oo, leftMargin) - // and the rightMargin which contains all the time partition on the unclosed right side: - // (rightMargin, +oo) - // all the remaining closed time range which locates in [leftMargin, rightMargin] will be - // calculated outside if block - long leftMargin = isNeedLeftAll ? partitionSlots.get(0).getStartTime() : Long.MIN_VALUE, - rightMargin = - isNeedRightAll - ? partitionSlots.get(partitionSlots.size() - 1).getStartTime() - : Long.MAX_VALUE; - seriesPartitionTable - .getSeriesPartitionMap() - .putAll( - seriesPartitionMap.entrySet().stream() - .filter( - entry -> { - long startTime = entry.getKey().getStartTime(); - return startTime < leftMargin || startTime > rightMargin; - }) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))); + try { + if (partitionSlots.isEmpty()) { + // Return all DataPartitions in one SeriesPartitionSlot + // when the queried TimePartitionSlots are empty + seriesPartitionTable.getSeriesPartitionMap().putAll(seriesPartitionMap); + } else { + 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 + // left side: (-oo, leftMargin) + // and the rightMargin which contains all the time partition on the unclosed right side: + // (rightMargin, +oo) + // all the remaining closed time range which locates in [leftMargin, rightMargin] will be + // calculated outside if block + long leftMargin = isNeedLeftAll ? partitionSlots.get(0).getStartTime() : Long.MIN_VALUE, + rightMargin = + isNeedRightAll + ? partitionSlots.get(partitionSlots.size() - 1).getStartTime() + : Long.MAX_VALUE; + seriesPartitionTable + .getSeriesPartitionMap() + .putAll( + seriesPartitionMap.entrySet().stream() + .filter( + entry -> { + long startTime = entry.getKey().getStartTime(); + return startTime < leftMargin || startTime > rightMargin; + }) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))); + } + + // Return the DataPartition for each match TimePartitionSlot + partitionSlots.forEach( + timePartitionSlot -> { + if (seriesPartitionMap.containsKey(timePartitionSlot)) { + seriesPartitionTable + .getSeriesPartitionMap() + .put(timePartitionSlot, seriesPartitionMap.get(timePartitionSlot)); + } else { + result.set(false); + } + }); } - - // Return the DataPartition for each match TimePartitionSlot - partitionSlots.forEach( - timePartitionSlot -> { - if (seriesPartitionMap.containsKey(timePartitionSlot)) { - seriesPartitionTable - .getSeriesPartitionMap() - .put(timePartitionSlot, seriesPartitionMap.get(timePartitionSlot)); - } else { - result.set(false); - } - }); + } finally { + seriesPartitionMapLock.readLock().unlock(); } - return result.get(); } @@ -130,8 +150,13 @@ public boolean getDataPartition( * @return The specified DataPartition's successor if exists, null otherwise */ public TConsensusGroupId getSuccessorDataPartition(TTimePartitionSlot timePartitionSlot) { - TTimePartitionSlot successorSlot = seriesPartitionMap.higherKey(timePartitionSlot); - return successorSlot == null ? null : seriesPartitionMap.get(successorSlot).get(0); + seriesPartitionMapLock.readLock().lock(); + try { + TTimePartitionSlot successorSlot = seriesPartitionMap.higherKey(timePartitionSlot); + return successorSlot == null ? null : seriesPartitionMap.get(successorSlot).get(0); + } finally { + seriesPartitionMapLock.readLock().unlock(); + } } /** @@ -141,8 +166,13 @@ public TConsensusGroupId getSuccessorDataPartition(TTimePartitionSlot timePartit * @return The specified DataPartition's predecessor if exists, null otherwise */ public TConsensusGroupId getPredecessorDataPartition(TTimePartitionSlot timePartitionSlot) { - TTimePartitionSlot predecessorSlot = seriesPartitionMap.lowerKey(timePartitionSlot); - return predecessorSlot == null ? null : seriesPartitionMap.get(predecessorSlot).get(0); + seriesPartitionMapLock.readLock().lock(); + try { + TTimePartitionSlot predecessorSlot = seriesPartitionMap.lowerKey(timePartitionSlot); + return predecessorSlot == null ? null : seriesPartitionMap.get(predecessorSlot).get(0); + } finally { + seriesPartitionMapLock.readLock().unlock(); + } } /** @@ -153,28 +183,38 @@ 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( - entry -> - entry.getKey().getStartTime() >= startTimeSlotId.getStartTime() - && entry.getKey().getStartTime() <= endTimeSlotId.getStartTime()) - .flatMap(entry -> entry.getValue().stream()) - .collect(Collectors.toList()); + seriesPartitionMapLock.readLock().lock(); + try { + return seriesPartitionMap.entrySet().stream() + .filter( + entry -> + entry.getKey().getStartTime() >= startTimeSlotId.getStartTime() + && entry.getKey().getStartTime() <= endTimeSlotId.getStartTime()) + .flatMap(entry -> entry.getValue().stream()) + .collect(Collectors.toList()); + } finally { + seriesPartitionMapLock.readLock().unlock(); + } } - List getTimeSlotList( + public List getTimeSlotList( TConsensusGroupId regionId, long startTime, long endTime) { - if (regionId.getId() == -1) { - return seriesPartitionMap.keySet().stream() - .filter(e -> e.getStartTime() >= startTime && e.getStartTime() < endTime) - .collect(Collectors.toList()); - } else { - return seriesPartitionMap.keySet().stream() - .filter(e -> e.getStartTime() >= startTime && e.getStartTime() < endTime) - .filter(e -> seriesPartitionMap.get(e).contains(regionId)) - .collect(Collectors.toList()); + seriesPartitionMapLock.readLock().lock(); + try { + if (regionId.getId() == -1) { + return seriesPartitionMap.keySet().stream() + .filter(e -> e.getStartTime() >= startTime && e.getStartTime() < endTime) + .collect(Collectors.toList()); + } else { + return seriesPartitionMap.keySet().stream() + .filter(e -> e.getStartTime() >= startTime && e.getStartTime() < endTime) + .filter(e -> seriesPartitionMap.get(e).contains(regionId)) + .collect(Collectors.toList()); + } + } finally { + seriesPartitionMapLock.readLock().unlock(); } } @@ -190,18 +230,23 @@ public void createDataPartition( SeriesPartitionTable assignedSeriesPartitionTable, TSeriesPartitionSlot seriesPartitionSlot, Map> groupDeltaMap) { - assignedSeriesPartitionTable - .getSeriesPartitionMap() - .forEach( - ((timePartitionSlot, consensusGroupIds) -> { - seriesPartitionMap.put(timePartitionSlot, new Vector<>(consensusGroupIds)); - consensusGroupIds.forEach( - consensusGroupId -> - groupDeltaMap - .computeIfAbsent(consensusGroupId, empty -> new ConcurrentHashMap<>()) - .computeIfAbsent(seriesPartitionSlot, empty -> new AtomicLong(0)) - .getAndIncrement()); - })); + seriesPartitionMapLock.writeLock().lock(); + try { + assignedSeriesPartitionTable + .getSeriesPartitionMap() + .forEach( + ((timePartitionSlot, consensusGroupIds) -> { + seriesPartitionMap.put(timePartitionSlot, new Vector<>(consensusGroupIds)); + consensusGroupIds.forEach( + consensusGroupId -> + groupDeltaMap + .computeIfAbsent(consensusGroupId, empty -> new ConcurrentHashMap<>()) + .computeIfAbsent(seriesPartitionSlot, empty -> new AtomicLong(0)) + .getAndIncrement()); + })); + } finally { + seriesPartitionMapLock.writeLock().unlock(); + } } /** @@ -213,15 +258,18 @@ public void createDataPartition( */ public synchronized List filterUnassignedDataPartitionSlots( List partitionSlots) { + seriesPartitionMapLock.readLock().lock(); List result = new Vector<>(); - - partitionSlots.forEach( - timePartitionSlot -> { - if (!seriesPartitionMap.containsKey(timePartitionSlot)) { - result.add(timePartitionSlot); - } - }); - + try { + partitionSlots.forEach( + timePartitionSlot -> { + if (!seriesPartitionMap.containsKey(timePartitionSlot)) { + result.add(timePartitionSlot); + } + }); + } finally { + seriesPartitionMapLock.readLock().unlock(); + } return result; } @@ -231,74 +279,100 @@ public synchronized List filterUnassignedDataPartitionSlots( * @return The last DataPartition's ConsensusGroupId, null if there are no DataPartitions yet */ public TConsensusGroupId getLastConsensusGroupId() { - Map.Entry> lastEntry = - seriesPartitionMap.lastEntry(); - if (lastEntry == null) { - return null; + seriesPartitionMapLock.readLock().lock(); + try { + Map.Entry> lastEntry = + seriesPartitionMap.lastEntry(); + if (lastEntry == null) { + return null; + } + return lastEntry.getValue().get(lastEntry.getValue().size() - 1); + } finally { + seriesPartitionMapLock.readLock().unlock(); } - return lastEntry.getValue().get(lastEntry.getValue().size() - 1); } public void serialize(OutputStream outputStream, TProtocol protocol) throws IOException, TException { - ReadWriteIOUtils.write(seriesPartitionMap.size(), outputStream); - for (Map.Entry> seriesPartitionEntry : - seriesPartitionMap.entrySet()) { - seriesPartitionEntry.getKey().write(protocol); - ReadWriteIOUtils.write(seriesPartitionEntry.getValue().size(), outputStream); - for (TConsensusGroupId consensusGroupId : seriesPartitionEntry.getValue()) { - consensusGroupId.write(protocol); + seriesPartitionMapLock.readLock().lock(); + try { + ReadWriteIOUtils.write(seriesPartitionMap.size(), outputStream); + for (Map.Entry> seriesPartitionEntry : + seriesPartitionMap.entrySet()) { + seriesPartitionEntry.getKey().write(protocol); + ReadWriteIOUtils.write(seriesPartitionEntry.getValue().size(), outputStream); + for (TConsensusGroupId consensusGroupId : seriesPartitionEntry.getValue()) { + consensusGroupId.write(protocol); + } } + } finally { + seriesPartitionMapLock.readLock().unlock(); } } /** Only for ConsensusRequest. */ public void deserialize(ByteBuffer buffer) { - int timePartitionSlotNum = buffer.getInt(); - 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)); + seriesPartitionMapLock.writeLock().lock(); + try { + int timePartitionSlotNum = buffer.getInt(); + 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); } - - seriesPartitionMap.put(timePartitionSlot, consensusGroupIds); + } finally { + seriesPartitionMapLock.writeLock().unlock(); } } /** Only for Snapshot. */ public void deserialize(InputStream inputStream, TProtocol protocol) throws IOException, TException { - int timePartitionSlotNum = ReadWriteIOUtils.readInt(inputStream); - 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++) { - TConsensusGroupId consensusGroupId = new TConsensusGroupId(); - consensusGroupId.read(protocol); - consensusGroupIds.add(consensusGroupId); + seriesPartitionMapLock.writeLock().lock(); + try { + int timePartitionSlotNum = ReadWriteIOUtils.readInt(inputStream); + 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++) { + TConsensusGroupId consensusGroupId = new TConsensusGroupId(); + consensusGroupId.read(protocol); + consensusGroupIds.add(consensusGroupId); + } + seriesPartitionMap.put(timePartitionSlot, consensusGroupIds); } - - seriesPartitionMap.put(timePartitionSlot, consensusGroupIds); + } finally { + seriesPartitionMapLock.writeLock().unlock(); } } @Override public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - SeriesPartitionTable that = (SeriesPartitionTable) o; - return seriesPartitionMap.equals(that.seriesPartitionMap); + seriesPartitionMapLock.readLock().lock(); + try { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + SeriesPartitionTable that = (SeriesPartitionTable) o; + return seriesPartitionMap.equals(that.seriesPartitionMap); + } finally { + seriesPartitionMapLock.readLock().unlock(); + } } @Override public int hashCode() { - return Objects.hash(seriesPartitionMap); + seriesPartitionMapLock.readLock().lock(); + try { + return Objects.hash(seriesPartitionMap); + } finally { + seriesPartitionMapLock.readLock().unlock(); + } } } From 94e95e8efeafdf7eee6fe573d8cf86112d3eebcd Mon Sep 17 00:00:00 2001 From: Yongzao Date: Fri, 7 Jun 2024 11:26:11 +0800 Subject: [PATCH 2/3] use concurrentskiplist --- .../partition/SeriesPartitionTable.java | 326 +++++++----------- 1 file changed, 123 insertions(+), 203 deletions(-) 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 b722f564b191b..9f773c37a122a 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 @@ -40,44 +40,30 @@ 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.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; public class SeriesPartitionTable { - private final ReentrantReadWriteLock seriesPartitionMapLock; - private final TreeMap> seriesPartitionMap; + private final ConcurrentSkipListMap> + seriesPartitionMap; public SeriesPartitionTable() { - this.seriesPartitionMapLock = new ReentrantReadWriteLock(); - this.seriesPartitionMap = new TreeMap<>(); + this.seriesPartitionMap = new ConcurrentSkipListMap<>(); } public SeriesPartitionTable(Map> seriesPartitionMap) { - this.seriesPartitionMapLock = new ReentrantReadWriteLock(); - this.seriesPartitionMap = new TreeMap<>(seriesPartitionMap); + this.seriesPartitionMap = new ConcurrentSkipListMap<>(seriesPartitionMap); } public Map> getSeriesPartitionMap() { - seriesPartitionMapLock.readLock().lock(); - try { - return seriesPartitionMap; - } finally { - seriesPartitionMapLock.readLock().unlock(); - } + return new TreeMap<>(seriesPartitionMap); } public void putDataPartition(TTimePartitionSlot timePartitionSlot, TConsensusGroupId groupId) { - seriesPartitionMapLock.writeLock().lock(); - try { - seriesPartitionMap - .computeIfAbsent(timePartitionSlot, empty -> new ArrayList<>()) - .add(groupId); - } finally { - seriesPartitionMapLock.writeLock().unlock(); - } + seriesPartitionMap.computeIfAbsent(timePartitionSlot, empty -> new ArrayList<>()).add(groupId); } /** @@ -90,55 +76,49 @@ public void putDataPartition(TTimePartitionSlot timePartitionSlot, TConsensusGro public boolean getDataPartition( TTimeSlotList partitionSlotList, SeriesPartitionTable seriesPartitionTable) { AtomicBoolean result = new AtomicBoolean(true); - seriesPartitionMapLock.readLock().lock(); List partitionSlots = partitionSlotList.getTimePartitionSlots(); - try { - if (partitionSlots.isEmpty()) { - // Return all DataPartitions in one SeriesPartitionSlot - // when the queried TimePartitionSlots are empty - seriesPartitionTable.getSeriesPartitionMap().putAll(seriesPartitionMap); - } else { - 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 - // left side: (-oo, leftMargin) - // and the rightMargin which contains all the time partition on the unclosed right side: - // (rightMargin, +oo) - // all the remaining closed time range which locates in [leftMargin, rightMargin] will be - // calculated outside if block - long leftMargin = isNeedLeftAll ? partitionSlots.get(0).getStartTime() : Long.MIN_VALUE, - rightMargin = - isNeedRightAll - ? partitionSlots.get(partitionSlots.size() - 1).getStartTime() - : Long.MAX_VALUE; - seriesPartitionTable - .getSeriesPartitionMap() - .putAll( - seriesPartitionMap.entrySet().stream() - .filter( - entry -> { - long startTime = entry.getKey().getStartTime(); - return startTime < leftMargin || startTime > rightMargin; - }) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))); - } - - // Return the DataPartition for each match TimePartitionSlot - partitionSlots.forEach( - timePartitionSlot -> { - if (seriesPartitionMap.containsKey(timePartitionSlot)) { - seriesPartitionTable - .getSeriesPartitionMap() - .put(timePartitionSlot, seriesPartitionMap.get(timePartitionSlot)); - } else { - result.set(false); - } - }); + if (partitionSlots.isEmpty()) { + // Return all DataPartitions in one SeriesPartitionSlot + // when the queried TimePartitionSlots are empty + seriesPartitionTable.getSeriesPartitionMap().putAll(seriesPartitionMap); + } else { + 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 + // left side: (-oo, leftMargin) + // and the rightMargin which contains all the time partition on the unclosed right side: + // (rightMargin, +oo) + // all the remaining closed time range which locates in [leftMargin, rightMargin] will be + // calculated outside if block + long leftMargin = isNeedLeftAll ? partitionSlots.get(0).getStartTime() : Long.MIN_VALUE, + rightMargin = + isNeedRightAll + ? partitionSlots.get(partitionSlots.size() - 1).getStartTime() + : Long.MAX_VALUE; + seriesPartitionTable + .getSeriesPartitionMap() + .putAll( + seriesPartitionMap.entrySet().stream() + .filter( + entry -> { + long startTime = entry.getKey().getStartTime(); + return startTime < leftMargin || startTime > rightMargin; + }) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))); } - } finally { - seriesPartitionMapLock.readLock().unlock(); + // Return the DataPartition for each match TimePartitionSlot + partitionSlots.forEach( + timePartitionSlot -> { + if (seriesPartitionMap.containsKey(timePartitionSlot)) { + seriesPartitionTable + .getSeriesPartitionMap() + .put(timePartitionSlot, seriesPartitionMap.get(timePartitionSlot)); + } else { + result.set(false); + } + }); } return result.get(); } @@ -150,13 +130,8 @@ public boolean getDataPartition( * @return The specified DataPartition's successor if exists, null otherwise */ public TConsensusGroupId getSuccessorDataPartition(TTimePartitionSlot timePartitionSlot) { - seriesPartitionMapLock.readLock().lock(); - try { - TTimePartitionSlot successorSlot = seriesPartitionMap.higherKey(timePartitionSlot); - return successorSlot == null ? null : seriesPartitionMap.get(successorSlot).get(0); - } finally { - seriesPartitionMapLock.readLock().unlock(); - } + TTimePartitionSlot successorSlot = seriesPartitionMap.higherKey(timePartitionSlot); + return successorSlot == null ? null : seriesPartitionMap.get(successorSlot).get(0); } /** @@ -166,13 +141,8 @@ public TConsensusGroupId getSuccessorDataPartition(TTimePartitionSlot timePartit * @return The specified DataPartition's predecessor if exists, null otherwise */ public TConsensusGroupId getPredecessorDataPartition(TTimePartitionSlot timePartitionSlot) { - seriesPartitionMapLock.readLock().lock(); - try { - TTimePartitionSlot predecessorSlot = seriesPartitionMap.lowerKey(timePartitionSlot); - return predecessorSlot == null ? null : seriesPartitionMap.get(predecessorSlot).get(0); - } finally { - seriesPartitionMapLock.readLock().unlock(); - } + TTimePartitionSlot predecessorSlot = seriesPartitionMap.lowerKey(timePartitionSlot); + return predecessorSlot == null ? null : seriesPartitionMap.get(predecessorSlot).get(0); } /** @@ -185,36 +155,26 @@ public TConsensusGroupId getPredecessorDataPartition(TTimePartitionSlot timePart */ public List getRegionId( TTimePartitionSlot startTimeSlotId, TTimePartitionSlot endTimeSlotId) { - seriesPartitionMapLock.readLock().lock(); - try { - return seriesPartitionMap.entrySet().stream() - .filter( - entry -> - entry.getKey().getStartTime() >= startTimeSlotId.getStartTime() - && entry.getKey().getStartTime() <= endTimeSlotId.getStartTime()) - .flatMap(entry -> entry.getValue().stream()) - .collect(Collectors.toList()); - } finally { - seriesPartitionMapLock.readLock().unlock(); - } + return seriesPartitionMap.entrySet().stream() + .filter( + entry -> + entry.getKey().getStartTime() >= startTimeSlotId.getStartTime() + && entry.getKey().getStartTime() <= endTimeSlotId.getStartTime()) + .flatMap(entry -> entry.getValue().stream()) + .collect(Collectors.toList()); } public List getTimeSlotList( TConsensusGroupId regionId, long startTime, long endTime) { - seriesPartitionMapLock.readLock().lock(); - try { - if (regionId.getId() == -1) { - return seriesPartitionMap.keySet().stream() - .filter(e -> e.getStartTime() >= startTime && e.getStartTime() < endTime) - .collect(Collectors.toList()); - } else { - return seriesPartitionMap.keySet().stream() - .filter(e -> e.getStartTime() >= startTime && e.getStartTime() < endTime) - .filter(e -> seriesPartitionMap.get(e).contains(regionId)) - .collect(Collectors.toList()); - } - } finally { - seriesPartitionMapLock.readLock().unlock(); + if (regionId.getId() == -1) { + return seriesPartitionMap.keySet().stream() + .filter(e -> e.getStartTime() >= startTime && e.getStartTime() < endTime) + .collect(Collectors.toList()); + } else { + return seriesPartitionMap.keySet().stream() + .filter(e -> e.getStartTime() >= startTime && e.getStartTime() < endTime) + .filter(e -> seriesPartitionMap.get(e).contains(regionId)) + .collect(Collectors.toList()); } } @@ -230,23 +190,18 @@ public void createDataPartition( SeriesPartitionTable assignedSeriesPartitionTable, TSeriesPartitionSlot seriesPartitionSlot, Map> groupDeltaMap) { - seriesPartitionMapLock.writeLock().lock(); - try { - assignedSeriesPartitionTable - .getSeriesPartitionMap() - .forEach( - ((timePartitionSlot, consensusGroupIds) -> { - seriesPartitionMap.put(timePartitionSlot, new Vector<>(consensusGroupIds)); - consensusGroupIds.forEach( - consensusGroupId -> - groupDeltaMap - .computeIfAbsent(consensusGroupId, empty -> new ConcurrentHashMap<>()) - .computeIfAbsent(seriesPartitionSlot, empty -> new AtomicLong(0)) - .getAndIncrement()); - })); - } finally { - seriesPartitionMapLock.writeLock().unlock(); - } + assignedSeriesPartitionTable + .getSeriesPartitionMap() + .forEach( + ((timePartitionSlot, consensusGroupIds) -> { + seriesPartitionMap.put(timePartitionSlot, new Vector<>(consensusGroupIds)); + consensusGroupIds.forEach( + consensusGroupId -> + groupDeltaMap + .computeIfAbsent(consensusGroupId, empty -> new ConcurrentHashMap<>()) + .computeIfAbsent(seriesPartitionSlot, empty -> new AtomicLong(0)) + .getAndIncrement()); + })); } /** @@ -258,18 +213,13 @@ public void createDataPartition( */ public synchronized List filterUnassignedDataPartitionSlots( List partitionSlots) { - seriesPartitionMapLock.readLock().lock(); List result = new Vector<>(); - try { - partitionSlots.forEach( - timePartitionSlot -> { - if (!seriesPartitionMap.containsKey(timePartitionSlot)) { - result.add(timePartitionSlot); - } - }); - } finally { - seriesPartitionMapLock.readLock().unlock(); - } + partitionSlots.forEach( + timePartitionSlot -> { + if (!seriesPartitionMap.containsKey(timePartitionSlot)) { + result.add(timePartitionSlot); + } + }); return result; } @@ -279,100 +229,70 @@ public synchronized List filterUnassignedDataPartitionSlots( * @return The last DataPartition's ConsensusGroupId, null if there are no DataPartitions yet */ public TConsensusGroupId getLastConsensusGroupId() { - seriesPartitionMapLock.readLock().lock(); - try { - Map.Entry> lastEntry = - seriesPartitionMap.lastEntry(); - if (lastEntry == null) { - return null; - } - return lastEntry.getValue().get(lastEntry.getValue().size() - 1); - } finally { - seriesPartitionMapLock.readLock().unlock(); + Map.Entry> lastEntry = + seriesPartitionMap.lastEntry(); + if (lastEntry == null) { + return null; } + return lastEntry.getValue().get(lastEntry.getValue().size() - 1); } public void serialize(OutputStream outputStream, TProtocol protocol) throws IOException, TException { - seriesPartitionMapLock.readLock().lock(); - try { - ReadWriteIOUtils.write(seriesPartitionMap.size(), outputStream); - for (Map.Entry> seriesPartitionEntry : - seriesPartitionMap.entrySet()) { - seriesPartitionEntry.getKey().write(protocol); - ReadWriteIOUtils.write(seriesPartitionEntry.getValue().size(), outputStream); - for (TConsensusGroupId consensusGroupId : seriesPartitionEntry.getValue()) { - consensusGroupId.write(protocol); - } + ReadWriteIOUtils.write(seriesPartitionMap.size(), outputStream); + for (Map.Entry> seriesPartitionEntry : + seriesPartitionMap.entrySet()) { + seriesPartitionEntry.getKey().write(protocol); + ReadWriteIOUtils.write(seriesPartitionEntry.getValue().size(), outputStream); + for (TConsensusGroupId consensusGroupId : seriesPartitionEntry.getValue()) { + consensusGroupId.write(protocol); } - } finally { - seriesPartitionMapLock.readLock().unlock(); } } /** Only for ConsensusRequest. */ public void deserialize(ByteBuffer buffer) { - seriesPartitionMapLock.writeLock().lock(); - try { - int timePartitionSlotNum = buffer.getInt(); - 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); + int timePartitionSlotNum = buffer.getInt(); + 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)); } - } finally { - seriesPartitionMapLock.writeLock().unlock(); + seriesPartitionMap.put(timePartitionSlot, consensusGroupIds); } } /** Only for Snapshot. */ public void deserialize(InputStream inputStream, TProtocol protocol) throws IOException, TException { - seriesPartitionMapLock.writeLock().lock(); - try { - int timePartitionSlotNum = ReadWriteIOUtils.readInt(inputStream); - 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++) { - TConsensusGroupId consensusGroupId = new TConsensusGroupId(); - consensusGroupId.read(protocol); - consensusGroupIds.add(consensusGroupId); - } - seriesPartitionMap.put(timePartitionSlot, consensusGroupIds); + int timePartitionSlotNum = ReadWriteIOUtils.readInt(inputStream); + 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++) { + TConsensusGroupId consensusGroupId = new TConsensusGroupId(); + consensusGroupId.read(protocol); + consensusGroupIds.add(consensusGroupId); } - } finally { - seriesPartitionMapLock.writeLock().unlock(); + seriesPartitionMap.put(timePartitionSlot, consensusGroupIds); } } @Override public boolean equals(Object o) { - seriesPartitionMapLock.readLock().lock(); - try { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - SeriesPartitionTable that = (SeriesPartitionTable) o; - return seriesPartitionMap.equals(that.seriesPartitionMap); - } finally { - seriesPartitionMapLock.readLock().unlock(); - } + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + SeriesPartitionTable that = (SeriesPartitionTable) o; + return seriesPartitionMap.equals(that.seriesPartitionMap); } @Override public int hashCode() { - seriesPartitionMapLock.readLock().lock(); - try { - return Objects.hash(seriesPartitionMap); - } finally { - seriesPartitionMapLock.readLock().unlock(); - } + return Objects.hash(seriesPartitionMap); } } From dd207eec9a8ecc07b614e868f1d61a36ab866718 Mon Sep 17 00:00:00 2001 From: Yongzao Date: Fri, 7 Jun 2024 16:55:38 +0800 Subject: [PATCH 3/3] bug fix --- .mvn/develocity.xml | 4 +--- .../iotdb/commons/partition/SeriesPartitionTable.java | 6 ++---- 2 files changed, 3 insertions(+), 7 deletions(-) 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/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 9f773c37a122a..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,11 +33,9 @@ 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; @@ -59,11 +57,11 @@ public SeriesPartitionTable(Map> ser } public Map> getSeriesPartitionMap() { - return new TreeMap<>(seriesPartitionMap); + return seriesPartitionMap; } public void putDataPartition(TTimePartitionSlot timePartitionSlot, TConsensusGroupId groupId) { - seriesPartitionMap.computeIfAbsent(timePartitionSlot, empty -> new ArrayList<>()).add(groupId); + seriesPartitionMap.computeIfAbsent(timePartitionSlot, empty -> new Vector<>()).add(groupId); } /**