Skip to content

Commit a498d7d

Browse files
CaideyipiCaideyipi
authored
Fix schema mutation validation, accounting and deletion performance (#18861)
* Fix schema mutation validation, accounting and deletion performance Validate aligned batch aliases, keep blacklist statistics idempotent, avoid repeated child scans during deletion, and correct and reuse device attribute storage. Add regression coverage for retries, snapshots, nested devices and all schema engine modes. * Fix aligned memchunk tests for current API
1 parent 0a79a49 commit a498d7d

10 files changed

Lines changed: 575 additions & 61 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/DeviceBlackListConstructor.java‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -64,9 +64,8 @@ public DeviceBlackListConstructor(
6464

6565
public void handleDeviceNode(final IDeviceMNode<IMemMNode> node) throws MetadataException {
6666
if (withoutFilter()) {
67-
node.preDeactivateSelfOrTemplate();
67+
preDeleteDevice(node);
6868
preDeletedNum++;
69-
regionStatistics.decreaseTableDevice(tableName, 1);
7069
return;
7170
}
7271
nodes.add(node);
@@ -76,14 +75,21 @@ public void handleDeviceNode(final IDeviceMNode<IMemMNode> node) throws Metadata
7675
@Override
7776
protected void update() {
7877
preDeletedNum += indexes.size();
79-
regionStatistics.decreaseTableDevice(tableName, indexes.size());
8078
for (final Integer index : indexes) {
81-
nodes.get(index).preDeactivateSelfOrTemplate();
79+
preDeleteDevice(nodes.get(index));
8280
}
8381
nodes.clear();
8482
super.clear();
8583
}
8684

85+
private void preDeleteDevice(final IDeviceMNode<IMemMNode> node) {
86+
// Retried requests still count matches, but statistics only change on the first transition.
87+
if (!node.isPreDeactivateSelfOrTemplate()) {
88+
node.preDeactivateSelfOrTemplate();
89+
regionStatistics.decreaseTableDevice(tableName, 1);
90+
}
91+
}
92+
8793
public long getPreDeletedNum() {
8894
return preDeletedNum;
8995
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/DeviceAttributeStore.java‎

Lines changed: 54 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import java.io.OutputStream;
4444
import java.nio.file.Files;
4545
import java.util.ArrayList;
46+
import java.util.BitSet;
4647
import java.util.HashMap;
4748
import java.util.List;
4849
import java.util.Map;
@@ -57,6 +58,9 @@ public class DeviceAttributeStore implements IDeviceAttributeStore {
5758
private static final long MAP_SIZE = RamUsageEstimator.shallowSizeOfInstance(HashMap.class);
5859

5960
private final List<Map<String, Binary>> deviceAttributeList = new ArrayList<>();
61+
// Reuse holes without moving live pointers or changing the snapshot format.
62+
private BitSet freeAttributePointers = new BitSet(0);
63+
private int nextFreeAttributePointer;
6064

6165
private final MemSchemaRegionStatistics regionStatistics;
6266

@@ -67,8 +71,10 @@ public DeviceAttributeStore(final MemSchemaRegionStatistics regionStatistics) {
6771
// It's OK not to calculate the region statistics here. The caller shall ensure that the
6872
// statistics is cleared at the outside.
6973
@Override
70-
public void clear() {
74+
public synchronized void clear() {
7175
deviceAttributeList.clear();
76+
freeAttributePointers = new BitSet(0);
77+
nextFreeAttributePointer = 0;
7278
}
7379

7480
@Override
@@ -113,7 +119,7 @@ public synchronized boolean createSnapshot(final File targetDir) {
113119
}
114120

115121
@Override
116-
public void loadFromSnapshot(final File snapshotDir) throws IOException {
122+
public synchronized void loadFromSnapshot(final File snapshotDir) throws IOException {
117123
final File snapshot =
118124
SystemFileFactory.INSTANCE.getFile(snapshotDir, SchemaConstant.DEVICE_ATTRIBUTE_SNAPSHOT);
119125
if (!snapshot.exists()) {
@@ -133,7 +139,7 @@ public void loadFromSnapshot(final File snapshotDir) throws IOException {
133139
public synchronized int createAttribute(
134140
final List<String> nameList, final Object[] valueList, final String tableName) {
135141
// todo implement storage for device of diverse data types
136-
long memUsage = MAP_SIZE + RamUsageEstimator.NUM_BYTES_OBJECT_REF;
142+
long memUsage = MAP_SIZE;
137143
final Map<String, Binary> attributeMap = new HashMap<>();
138144
for (int i = 0; i < nameList.size(); i++) {
139145
if (valueList.length <= i) {
@@ -147,19 +153,29 @@ public synchronized int createAttribute(
147153
memUsage += MemUsageUtil.computeKVMemUsageInMap(nameList.get(i), value);
148154
addTableAttributeMemory(tableName, value.ramBytesUsed());
149155
}
150-
deviceAttributeList.add(attributeMap);
156+
int pointer = freeAttributePointers.nextSetBit(nextFreeAttributePointer);
157+
if (pointer < 0) {
158+
pointer = deviceAttributeList.size();
159+
deviceAttributeList.add(attributeMap);
160+
memUsage += RamUsageEstimator.NUM_BYTES_OBJECT_REF;
161+
} else {
162+
deviceAttributeList.set(pointer, attributeMap);
163+
freeAttributePointers.clear(pointer);
164+
nextFreeAttributePointer = pointer + 1;
165+
}
151166
requestMemory(memUsage);
152-
return deviceAttributeList.size() - 1;
167+
return pointer;
153168
}
154169

155170
@Override
156-
public Map<String, Binary> alterAttribute(
171+
public synchronized Map<String, Binary> alterAttribute(
157172
final int pointer,
158173
final List<String> nameList,
159174
final Object[] valueList,
160175
final String tableName) {
161176
// todo implement storage for device of diverse data types
162177
long memUsageDelta = 0L;
178+
long valueMemUsageDelta = 0L;
163179
long originMemUsage;
164180
long updatedMemUsage;
165181
final Map<String, Binary> updateMap = new HashMap<>();
@@ -173,11 +189,12 @@ public Map<String, Binary> alterAttribute(
173189
continue;
174190
}
175191
final Binary value = (Binary) valueList[i];
192+
final Binary oldValue = attributeMap.get(key);
193+
valueMemUsageDelta +=
194+
(value == null ? 0 : value.ramBytesUsed())
195+
- (oldValue == null ? 0 : oldValue.ramBytesUsed());
176196

177-
originMemUsage =
178-
attributeMap.containsKey(key)
179-
? MemUsageUtil.computeKVMemUsageInMap(key, attributeMap.get(key))
180-
: 0;
197+
originMemUsage = oldValue == null ? 0 : MemUsageUtil.computeKVMemUsageInMap(key, oldValue);
181198
if (value != null) {
182199
if (!Objects.equals(value, attributeMap.put(key, value))) {
183200
updateMap.put(key, value);
@@ -193,28 +210,33 @@ public Map<String, Binary> alterAttribute(
193210
}
194211
if (memUsageDelta > 0) {
195212
requestMemory(memUsageDelta);
196-
addTableAttributeMemory(tableName, memUsageDelta);
197213
} else if (memUsageDelta < 0) {
198214
releaseMemory(-memUsageDelta);
199-
decreaseTableAttributeMemory(tableName, -memUsageDelta);
215+
}
216+
if (valueMemUsageDelta > 0) {
217+
addTableAttributeMemory(tableName, valueMemUsageDelta);
218+
} else if (valueMemUsageDelta < 0) {
219+
decreaseTableAttributeMemory(tableName, -valueMemUsageDelta);
200220
}
201221
return updateMap;
202222
}
203223

204224
@Override
205-
public void removeAttribute(final int pointer, final String tableName) {
206-
releaseMemory(
207-
MAP_SIZE + UpdateDetailContainer.sizeOfMapEntries(deviceAttributeList.get(pointer)));
225+
public synchronized void removeAttribute(final int pointer, final String tableName) {
226+
final Map<String, Binary> attributeMap = deviceAttributeList.get(pointer);
227+
if (attributeMap == null) {
228+
return;
229+
}
230+
releaseMemory(MAP_SIZE + UpdateDetailContainer.sizeOfMapEntries(attributeMap));
208231
decreaseTableAttributeMemory(
209232
tableName,
210-
deviceAttributeList.get(pointer).values().stream()
211-
.map(UpdateDetailContainer::sizeOf)
212-
.reduce(0L, Long::sum));
233+
attributeMap.values().stream().map(UpdateDetailContainer::sizeOf).reduce(0L, Long::sum));
213234
deviceAttributeList.set(pointer, null);
235+
markFreeAttributePointer(pointer);
214236
}
215237

216238
@Override
217-
public void removeAttribute(
239+
public synchronized void removeAttribute(
218240
final int pointer, final String attributeName, final String tableName) {
219241
final Map<String, Binary> attributeMap = deviceAttributeList.get(pointer);
220242
if (Objects.isNull(attributeMap)) {
@@ -228,12 +250,12 @@ public void removeAttribute(
228250
}
229251

230252
@Override
231-
public Map<String, Binary> getAttributes(final int pointer) {
253+
public synchronized Map<String, Binary> getAttributes(final int pointer) {
232254
return deviceAttributeList.get(pointer);
233255
}
234256

235257
@Override
236-
public Binary getAttributes(final int pointer, final String name) {
258+
public synchronized Binary getAttributes(final int pointer, final String name) {
237259
return deviceAttributeList.get(pointer).get(name);
238260
}
239261

@@ -271,6 +293,9 @@ private void deserialize(final InputStream inputStream) throws IOException {
271293
for (int i = 0; i < size; i++) {
272294
final Map<String, Binary> attributeMap = readMap(inputStream, false);
273295
deviceAttributeList.add(attributeMap);
296+
if (attributeMap == null) {
297+
markFreeAttributePointer(i);
298+
}
274299
requestMemory(
275300
RamUsageEstimator.NUM_BYTES_OBJECT_REF
276301
+ (Objects.nonNull(attributeMap)
@@ -279,6 +304,14 @@ private void deserialize(final InputStream inputStream) throws IOException {
279304
}
280305
}
281306

307+
private void markFreeAttributePointer(final int pointer) {
308+
final int oldCapacity = freeAttributePointers.size();
309+
freeAttributePointers.set(pointer);
310+
// Account for bitmap storage as well as the retained references in deviceAttributeList.
311+
requestMemory((freeAttributePointers.size() - oldCapacity) / Byte.SIZE);
312+
nextFreeAttributePointer = Math.min(nextFreeAttributePointer, pointer);
313+
}
314+
282315
public static Map<String, Binary> readMap(final InputStream inputStream, final boolean concurrent)
283316
throws IOException {
284317
final int length = ReadWriteIOUtils.readInt(inputStream);

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java‎

Lines changed: 5 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
import org.apache.iotdb.db.schemaengine.metric.SchemaRegionMemMetric;
5555
import org.apache.iotdb.db.schemaengine.rescon.MemSchemaRegionStatistics;
5656
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.IMemMNode;
57+
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.impl.BasicInternalMNode;
5758
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.TableDeviceInfo;
5859
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.loader.MNodeFactoryLoader;
5960
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.traverser.collector.EntityCollector;
@@ -403,6 +404,7 @@ public List<IMeasurementMNode<IMemMNode>> createAlignedTimeSeries(
403404
throws MetadataException {
404405
final List<IMeasurementMNode<IMemMNode>> measurementMNodeList = new ArrayList<>();
405406
MetaFormatUtils.checkSchemaMeasurementNames(measurements);
407+
MetaFormatUtils.checkSchemaMeasurementAliases(devicePath, measurements, aliasList);
406408
final IMemMNode deviceParent = checkAndAutoCreateInternalPath(devicePath);
407409

408410
// synchronize check and add, we need addChild operation be atomic.
@@ -706,28 +708,13 @@ public IMeasurementMNode<IMemMNode> deleteTimeSeries(final PartialPath path)
706708
public void deleteEmptyInternalMNode(final IDeviceMNode<IMemMNode> entityMNode) {
707709
IMemMNode curNode = entityMNode.getAsMNode();
708710
if (!entityMNode.isUseTemplate()) {
709-
boolean hasMeasurement = false;
710-
boolean hasNonViewMeasurement = false;
711-
IMemMNode child;
712-
try (final IMNodeIterator<IMemMNode> iterator = store.getChildrenIterator(curNode)) {
713-
while (iterator.hasNext()) {
714-
child = iterator.next();
715-
if (child.isMeasurement()) {
716-
hasMeasurement = true;
717-
if (!child.getAsMeasurementMNode().isLogicalView()) {
718-
hasNonViewMeasurement = true;
719-
break;
720-
}
721-
}
722-
}
723-
}
724-
725-
if (!hasMeasurement) {
711+
final BasicInternalMNode internalNode = (BasicInternalMNode) curNode.getBasicMNode();
712+
if (internalNode.getMeasurementChildCount() == 0) {
726713
synchronized (this) {
727714
curNode = store.setToInternal(entityMNode);
728715
}
729716
refreshAncestorsHavingDeviceDescendant(curNode.getParent());
730-
} else if (!hasNonViewMeasurement) {
717+
} else if (internalNode.getNonViewMeasurementChildCount() == 0) {
731718
// has some measurement but they are all logical view
732719
entityMNode.setAligned(null);
733720
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/BasicInternalMNode.java‎

Lines changed: 42 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,10 @@ public class BasicInternalMNode extends BasicMNode implements IInternalMNode<IMe
4747
@SuppressWarnings("squid:S3077")
4848
private volatile IDeviceInfo<IMemMNode> deviceInfo = null;
4949

50+
// Direct children only. Snapshot loading rebuilds these counters through addChild.
51+
private int measurementChildCount;
52+
private int nonViewMeasurementChildCount;
53+
5054
/** Constructor of MNode. */
5155
public BasicInternalMNode(IMemMNode parent, String name) {
5256
super(parent, name);
@@ -98,6 +102,9 @@ public IMemMNode addChild(String name, IMemMNode child) {
98102
}
99103
child.setParent(this);
100104
IMemMNode existingChild = children.putIfAbsent(name, child);
105+
if (existingChild == null) {
106+
updateMeasurementChildCount(child, 1);
107+
}
101108
return existingChild == null ? child : existingChild;
102109
}
103110

@@ -114,29 +121,19 @@ public IMemMNode addChild(String name, IMemMNode child) {
114121
*/
115122
@Override
116123
public IMemMNode addChild(IMemMNode child) {
117-
/* use cpu time to exchange memory
118-
* measurementNode's children should be null to save memory
119-
* add child method will only be called when writing MTree, which is not a frequent operation
120-
*/
121-
if (children == null) {
122-
// double check, children is volatile
123-
synchronized (this) {
124-
if (children == null) {
125-
children = new MemMNodeContainer();
126-
}
127-
}
128-
}
129-
130-
child.setParent(this);
131-
children.putIfAbsent(child.getName(), child);
124+
addChild(child.getName(), child);
132125
return child;
133126
}
134127

135128
/** Delete a child. */
136129
@Override
137130
public IMemMNode deleteChild(String name) {
138131
if (children != null) {
139-
return children.remove(name);
132+
final IMemMNode child = children.remove(name);
133+
if (child != null) {
134+
updateMeasurementChildCount(child, -1);
135+
}
136+
return child;
140137
}
141138
return null;
142139
}
@@ -152,12 +149,39 @@ public IMNodeContainer<IMemMNode> getChildren() {
152149
@Override
153150
public void setChildren(IMNodeContainer<IMemMNode> children) {
154151
this.children = children;
152+
measurementChildCount = 0;
153+
nonViewMeasurementChildCount = 0;
154+
if (children != null) {
155+
children.values().forEach(child -> updateMeasurementChildCount(child, 1));
156+
}
157+
}
158+
159+
private void updateMeasurementChildCount(final IMemMNode child, final int delta) {
160+
if (child.isMeasurement()) {
161+
measurementChildCount += delta;
162+
if (!child.getAsMeasurementMNode().isLogicalView()) {
163+
nonViewMeasurementChildCount += delta;
164+
}
165+
}
166+
}
167+
168+
public int getMeasurementChildCount() {
169+
return measurementChildCount;
170+
}
171+
172+
public int getNonViewMeasurementChildCount() {
173+
return nonViewMeasurementChildCount;
155174
}
156175

157-
/** MNodeContainer reference and basic occupation, 8 + 80B. DeviceInfo reference and size. */
176+
/** MNodeContainer reference and basic occupation, 8 + 80B. Counters, 8B. DeviceInfo and size. */
158177
@Override
159178
public int estimateSize() {
160-
return 8 + 80 + super.estimateSize() + 8 + (deviceInfo == null ? 0 : deviceInfo.estimateSize());
179+
return 8
180+
+ 80
181+
+ 8
182+
+ super.estimateSize()
183+
+ 8
184+
+ (deviceInfo == null ? 0 : deviceInfo.estimateSize());
161185
}
162186

163187
@Override

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -471,6 +471,7 @@ public List<IMeasurementMNode<ICachedMNode>> createAlignedTimeSeries(
471471
throws MetadataException {
472472
final List<IMeasurementMNode<ICachedMNode>> measurementMNodeList = new ArrayList<>();
473473
MetaFormatUtils.checkSchemaMeasurementNames(measurements);
474+
MetaFormatUtils.checkSchemaMeasurementAliases(devicePath, measurements, aliasList);
474475
final ICachedMNode deviceParent = checkAndAutoCreateInternalPath(devicePath);
475476

476477
try {

0 commit comments

Comments
 (0)