Skip to content

Commit 9b124e4

Browse files
CaideyipiCaideyipi
authored
Fix tag index memory accounting (#18029) (#18040)
* Fix tag index memory accounting * Add concurrent tag index memory test * Clarify tag index size accounting * Clarify tag index overhead estimate * Fix pipe source time range IT flakiness (cherry picked from commit 580fd6a)
1 parent a653aee commit 9b124e4

3 files changed

Lines changed: 308 additions & 64 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeSourceIT.java‎

Lines changed: 23 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@
5252
import java.util.HashSet;
5353
import java.util.List;
5454
import java.util.Map;
55+
import java.util.function.Consumer;
5556

5657
import static org.junit.Assert.fail;
5758

@@ -761,6 +762,11 @@ public void testExtractorTimeRangeMatch() throws Exception {
761762

762763
final String receiverIp = receiverDataNode.getIp();
763764
final int receiverPort = receiverDataNode.getPort();
765+
final Consumer<String> handleFailure =
766+
o -> {
767+
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
768+
TestUtils.executeNonQueryWithRetry(receiverEnv, "flush");
769+
};
764770

765771
try (final SyncConfigNodeIServiceClient client =
766772
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
@@ -795,10 +801,12 @@ public void testExtractorTimeRangeMatch() throws Exception {
795801
.setProcessorAttributes(processorAttributes));
796802
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
797803

798-
final Map<String, String> expectedCountResult = new HashMap<>();
799-
expectedCountResult.put("count(root.db.d1.at1)", "3");
800804
TestUtils.assertDataEventuallyOnEnv(
801-
receiverEnv, "select count(*) from root.db.**", expectedCountResult);
805+
receiverEnv,
806+
"select count(at1) from root.db.d1",
807+
"count(root.db.d1.at1),",
808+
Collections.singleton("3,"),
809+
handleFailure);
802810

803811
// Insert realtime data that overlapped with time range
804812
TestUtils.executeNonQueries(
@@ -809,9 +817,12 @@ public void testExtractorTimeRangeMatch() throws Exception {
809817
"flush"),
810818
null);
811819

812-
expectedCountResult.put("count(root.db.d3.at1)", "3");
813820
TestUtils.assertDataEventuallyOnEnv(
814-
receiverEnv, "select count(*) from root.db.**", expectedCountResult);
821+
receiverEnv,
822+
"select count(at1) from root.db.d1, root.db.d3",
823+
"count(root.db.d1.at1),count(root.db.d3.at1),",
824+
Collections.singleton("3,3,"),
825+
handleFailure);
815826

816827
// Session Tablet can have unused timestamp slots when rowSize is smaller than maxRowNumber.
817828
// The pipe source time range filter should ignore the unused zero tail.
@@ -829,9 +840,12 @@ public void testExtractorTimeRangeMatch() throws Exception {
829840
session.insertTablet(tabletWithUnusedTail);
830841
}
831842

832-
expectedCountResult.put("count(root.db.d5.at1)", "3");
833843
TestUtils.assertDataEventuallyOnEnv(
834-
receiverEnv, "select count(*) from root.db.**", expectedCountResult);
844+
receiverEnv,
845+
"select count(at1) from root.db.d1, root.db.d3, root.db.d5",
846+
"count(root.db.d1.at1),count(root.db.d3.at1),count(root.db.d5.at1),",
847+
Collections.singleton("3,3,3,"),
848+
handleFailure);
835849

836850
// Insert realtime data that does not overlap with time range
837851
TestUtils.executeNonQueries(
@@ -846,7 +860,8 @@ public void testExtractorTimeRangeMatch() throws Exception {
846860
receiverEnv,
847861
"select count(at1) from root.db.d1, root.db.d3, root.db.d5",
848862
"count(root.db.d1.at1),count(root.db.d3.at1),count(root.db.d5.at1),",
849-
Collections.singleton("3,3,3,"));
863+
Collections.singleton("3,3,3,"),
864+
600);
850865
TestUtils.assertDataAlwaysOnEnv(
851866
receiverEnv,
852867
"show timeseries root.db.d2.**",

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/tag/TagManager.java‎

Lines changed: 69 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,6 @@
5151
import java.util.Collections;
5252
import java.util.Comparator;
5353
import java.util.HashMap;
54-
import java.util.HashSet;
5554
import java.util.Iterator;
5655
import java.util.List;
5756
import java.util.Map;
@@ -70,6 +69,10 @@ public class TagManager {
7069
private static final String PREVIOUS_CONDITION =
7170
"before deleting it, tag key is %s, tag value is %s, tlog offset is %d, contains key %b";
7271

72+
// The tag index memory model adds one int-sized estimated overhead for each indexed key, value,
73+
// and measurement reference. This is an accounting estimate rather than a specific
74+
// ConcurrentHashMap or Set field.
75+
private static final long INDEX_ENTRY_OVERHEAD_ESTIMATE_BYTES = Integer.BYTES;
7376
private static final Logger logger = LoggerFactory.getLogger(TagManager.class);
7477
private static final CommonConfig COMMON_CONFIG = CommonDescriptor.getInstance().getConfig();
7578

@@ -164,34 +167,31 @@ public void addIndex(String tagKey, String tagValue, IMeasurementMNode<?> measur
164167
return;
165168
}
166169

167-
int tagIndexOldSize = tagIndex.size();
168-
Map<String, Set<IMeasurementMNode<?>>> tagValueMap =
169-
tagIndex.computeIfAbsent(tagKey, k -> new ConcurrentHashMap<>());
170-
int tagIndexNewSize = tagIndex.size();
171-
172-
int tagValueMapOldSize = tagValueMap.size();
173-
Set<IMeasurementMNode<?>> measurementsSet =
174-
tagValueMap.computeIfAbsent(tagValue, v -> Collections.synchronizedSet(new HashSet<>()));
175-
int tagValueMapNewSize = tagValueMap.size();
170+
tagIndex.compute(
171+
tagKey,
172+
(key, tagValueMap) -> {
173+
long memorySize = 0;
174+
if (tagValueMap == null) {
175+
tagValueMap = new ConcurrentHashMap<>();
176+
memorySize += RamUsageEstimator.sizeOf(tagKey) + INDEX_ENTRY_OVERHEAD_ESTIMATE_BYTES;
177+
}
176178

177-
int measurementsSetOldSize = measurementsSet.size();
178-
measurementsSet.add(measurementMNode);
179-
int measurementsSetNewSize = measurementsSet.size();
179+
Set<IMeasurementMNode<?>> measurementsSet = tagValueMap.get(tagValue);
180+
if (measurementsSet == null) {
181+
measurementsSet = ConcurrentHashMap.newKeySet();
182+
tagValueMap.put(tagValue, measurementsSet);
183+
memorySize += RamUsageEstimator.sizeOf(tagValue) + INDEX_ENTRY_OVERHEAD_ESTIMATE_BYTES;
184+
}
180185

181-
long memorySize = 0;
182-
if (tagIndexNewSize - tagIndexOldSize == 1) {
183-
// the last 4 is the memory occupied by the size of tagvaluemap
184-
memorySize += RamUsageEstimator.sizeOf(tagKey) + 4;
185-
}
186-
if (tagValueMapNewSize - tagValueMapOldSize == 1) {
187-
// the last 4 is the memory occupied by the size of measurementsSet
188-
memorySize += RamUsageEstimator.sizeOf(tagValue) + 4;
189-
}
190-
if (measurementsSetNewSize - measurementsSetOldSize == 1) {
191-
// 8 is the memory occupied by the length of the IMeasurementMNode
192-
memorySize += RamUsageEstimator.NUM_BYTES_OBJECT_REF + 4;
193-
}
194-
requestMemory(memorySize);
186+
if (measurementsSet.add(measurementMNode)) {
187+
memorySize +=
188+
RamUsageEstimator.NUM_BYTES_OBJECT_REF + INDEX_ENTRY_OVERHEAD_ESTIMATE_BYTES;
189+
}
190+
if (memorySize > 0) {
191+
requestMemory(memorySize);
192+
}
193+
return tagValueMap;
194+
});
195195
}
196196

197197
public void addIndex(Map<String, String> tagsMap, IMeasurementMNode<?> measurementMNode) {
@@ -206,32 +206,47 @@ public void removeIndex(String tagKey, String tagValue, IMeasurementMNode<?> mea
206206
if (tagKey == null || tagValue == null || measurementMNode == null) {
207207
return;
208208
}
209-
// init memory size
210-
long memorySize = 0;
211-
if (tagIndex.get(tagKey).get(tagValue).remove(measurementMNode)) {
212-
memorySize += RamUsageEstimator.NUM_BYTES_OBJECT_REF + 4;
213-
}
214-
if (tagIndex.get(tagKey).get(tagValue).isEmpty()) {
215-
if (tagIndex.get(tagKey).remove(tagValue) != null) {
216-
// the last 4 is the memory occupied by the size of IMeasurementMNodeSet
217-
memorySize += RamUsageEstimator.sizeOf(tagValue) + 4;
218-
}
219-
}
220-
if (tagIndex.get(tagKey).isEmpty()) {
221-
if (tagIndex.remove(tagKey) != null) {
222-
// the last 4 is the memory occupied by the size of tagValueMap
223-
memorySize += RamUsageEstimator.sizeOf(tagKey) + 4;
224-
}
225-
}
226-
releaseMemory(memorySize);
209+
tagIndex.computeIfPresent(
210+
tagKey,
211+
(key, tagValueMap) -> {
212+
long memorySize = 0;
213+
Set<IMeasurementMNode<?>> measurementsSet = tagValueMap.get(tagValue);
214+
if (measurementsSet == null) {
215+
return tagValueMap;
216+
}
217+
218+
if (measurementsSet.remove(measurementMNode)) {
219+
memorySize +=
220+
RamUsageEstimator.NUM_BYTES_OBJECT_REF + INDEX_ENTRY_OVERHEAD_ESTIMATE_BYTES;
221+
}
222+
if (measurementsSet.isEmpty()) {
223+
if (tagValueMap.remove(tagValue, measurementsSet)) {
224+
memorySize +=
225+
RamUsageEstimator.sizeOf(tagValue) + INDEX_ENTRY_OVERHEAD_ESTIMATE_BYTES;
226+
}
227+
}
228+
if (tagValueMap.isEmpty()) {
229+
memorySize += RamUsageEstimator.sizeOf(tagKey) + INDEX_ENTRY_OVERHEAD_ESTIMATE_BYTES;
230+
if (memorySize > 0) {
231+
releaseMemory(memorySize);
232+
}
233+
return null;
234+
}
235+
if (memorySize > 0) {
236+
releaseMemory(memorySize);
237+
}
238+
return tagValueMap;
239+
});
240+
}
241+
242+
private boolean containsIndex(String tagKey, String tagValue) {
243+
Map<String, Set<IMeasurementMNode<?>>> tagValueMap = tagIndex.get(tagKey);
244+
return tagValueMap != null && tagValueMap.containsKey(tagValue);
227245
}
228246

229247
private List<IMeasurementMNode<?>> getMatchedTimeseriesInIndex(TagFilter tagFilter) {
230-
if (!tagIndex.containsKey(tagFilter.getKey())) {
231-
return Collections.emptyList();
232-
}
233248
Map<String, Set<IMeasurementMNode<?>>> value2Node = tagIndex.get(tagFilter.getKey());
234-
if (value2Node.isEmpty()) {
249+
if (value2Node == null || value2Node.isEmpty()) {
235250
return Collections.emptyList();
236251
}
237252

@@ -362,8 +377,7 @@ public void removeFromTagInvertedIndex(IMeasurementMNode<?> node) throws IOExcep
362377
Map<String, String> tagMap = tagLogFile.readTag(node.getOffset());
363378
if (tagMap != null) {
364379
for (Map.Entry<String, String> entry : tagMap.entrySet()) {
365-
if (tagIndex.containsKey(entry.getKey())
366-
&& tagIndex.get(entry.getKey()).containsKey(entry.getValue())) {
380+
if (containsIndex(entry.getKey(), entry.getValue())) {
367381
if (logger.isDebugEnabled()) {
368382
logger.debug(
369383
String.format(
@@ -415,7 +429,7 @@ public void updateTagsAndAttributes(
415429
// we should remove before key-value from inverted index map
416430
if (beforeValue != null && !beforeValue.equals(value)) {
417431

418-
if (tagIndex.containsKey(key) && tagIndex.get(key).containsKey(beforeValue)) {
432+
if (containsIndex(key, beforeValue)) {
419433
if (logger.isDebugEnabled()) {
420434
logger.debug(
421435
String.format(
@@ -548,8 +562,7 @@ public void dropTagsOrAttributes(
548562

549563
if (!deleteTag.isEmpty()) {
550564
for (Map.Entry<String, String> entry : deleteTag.entrySet()) {
551-
if (tagIndex.containsKey((entry.getKey()))
552-
&& tagIndex.get(entry.getKey()).containsKey(entry.getValue())) {
565+
if (containsIndex(entry.getKey(), entry.getValue())) {
553566
if (logger.isDebugEnabled()) {
554567
logger.debug(
555568
String.format(
@@ -618,7 +631,7 @@ public void setTagsOrAttributesValue(
618631
String beforeValue = entry.getValue();
619632
String currentValue = newTagValue.get(key);
620633
// change the tag inverted index map
621-
if (tagIndex.containsKey(key) && tagIndex.get(key).containsKey(beforeValue)) {
634+
if (containsIndex(key, beforeValue)) {
622635

623636
if (logger.isDebugEnabled()) {
624637
logger.debug(
@@ -676,7 +689,7 @@ public void renameTagOrAttributeKey(
676689
// persist the change to disk
677690
tagLogFile.write(pair.left, pair.right, leafMNode.getOffset());
678691
// change the tag inverted index map
679-
if (tagIndex.containsKey(oldKey) && tagIndex.get(oldKey).containsKey(value)) {
692+
if (containsIndex(oldKey, value)) {
680693

681694
if (logger.isDebugEnabled()) {
682695
logger.debug(

0 commit comments

Comments
 (0)