Skip to content

Commit f6b122c

Browse files
CaideyipiCaideyipi
authored
Fix schema mutation validation, accounting and deletion performance (#18861) (#18871)
* 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 (cherry picked from commit a498d7d)
1 parent 6527fb0 commit f6b122c

6 files changed

Lines changed: 261 additions & 36 deletions

File tree

‎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
@@ -45,6 +45,7 @@
4545
import org.apache.iotdb.db.schemaengine.metric.SchemaRegionMemMetric;
4646
import org.apache.iotdb.db.schemaengine.rescon.MemSchemaRegionStatistics;
4747
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.IMemMNode;
48+
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.impl.BasicInternalMNode;
4849
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.loader.MNodeFactoryLoader;
4950
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.traverser.collector.EntityCollector;
5051
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.traverser.collector.MNodeCollector;
@@ -374,6 +375,7 @@ public List<IMeasurementMNode<IMemMNode>> createAlignedTimeSeries(
374375
throws MetadataException {
375376
final List<IMeasurementMNode<IMemMNode>> measurementMNodeList = new ArrayList<>();
376377
MetaFormatUtils.checkSchemaMeasurementNames(measurements);
378+
MetaFormatUtils.checkSchemaMeasurementAliases(devicePath, measurements, aliasList);
377379
final IMemMNode deviceParent = checkAndAutoCreateInternalPath(devicePath);
378380

379381
// synchronize check and add, we need addChild operation be atomic.
@@ -607,28 +609,13 @@ public IMeasurementMNode<IMemMNode> deleteTimeseries(PartialPath path) throws Me
607609
public void deleteEmptyInternalMNode(IDeviceMNode<IMemMNode> entityMNode) {
608610
IMemMNode curNode = entityMNode.getAsMNode();
609611
if (!entityMNode.isUseTemplate()) {
610-
boolean hasMeasurement = false;
611-
boolean hasNonViewMeasurement = false;
612-
IMemMNode child;
613-
try (final IMNodeIterator<IMemMNode> iterator = store.getChildrenIterator(curNode)) {
614-
while (iterator.hasNext()) {
615-
child = iterator.next();
616-
if (child.isMeasurement()) {
617-
hasMeasurement = true;
618-
if (!child.getAsMeasurementMNode().isLogicalView()) {
619-
hasNonViewMeasurement = true;
620-
break;
621-
}
622-
}
623-
}
624-
}
625-
626-
if (!hasMeasurement) {
612+
final BasicInternalMNode internalNode = (BasicInternalMNode) curNode.getBasicMNode();
613+
if (internalNode.getMeasurementChildCount() == 0) {
627614
synchronized (this) {
628615
curNode = store.setToInternal(entityMNode);
629616
}
630617
refreshAncestorsHavingDeviceDescendant(curNode.getParent());
631-
} else if (!hasNonViewMeasurement) {
618+
} else if (internalNode.getNonViewMeasurementChildCount() == 0) {
632619
// has some measurement but they are all logical view
633620
entityMNode.setAligned(null);
634621
}

‎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
@@ -46,6 +46,10 @@ public class BasicInternalMNode extends BasicMNode implements IInternalMNode<IMe
4646
@SuppressWarnings("squid:S3077")
4747
private volatile IDeviceInfo<IMemMNode> deviceInfo = null;
4848

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

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

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

156-
/** MNodeContainer reference and basic occupation, 8 + 80B. DeviceInfo reference and size. */
175+
/** MNodeContainer reference and basic occupation, 8 + 80B. Counters, 8B. DeviceInfo and size. */
157176
@Override
158177
public int estimateSize() {
159-
return 8 + 80 + super.estimateSize() + 8 + (deviceInfo == null ? 0 : deviceInfo.estimateSize());
178+
return 8
179+
+ 80
180+
+ 8
181+
+ super.estimateSize()
182+
+ 8
183+
+ (deviceInfo == null ? 0 : deviceInfo.estimateSize());
160184
}
161185

162186
@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
@@ -465,6 +465,7 @@ public List<IMeasurementMNode<ICachedMNode>> createAlignedTimeSeries(
465465
throws MetadataException {
466466
final List<IMeasurementMNode<ICachedMNode>> measurementMNodeList = new ArrayList<>();
467467
MetaFormatUtils.checkSchemaMeasurementNames(measurements);
468+
MetaFormatUtils.checkSchemaMeasurementAliases(devicePath, measurements, aliasList);
468469
final ICachedMNode deviceParent = checkAndAutoCreateInternalPath(devicePath);
469470

470471
try {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/MetaFormatUtils.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,14 +22,17 @@
2222
import org.apache.iotdb.commons.exception.MetadataException;
2323
import org.apache.iotdb.commons.path.PartialPath;
2424
import org.apache.iotdb.db.conf.IoTDBConfig;
25+
import org.apache.iotdb.db.exception.metadata.AliasAlreadyExistException;
2526
import org.apache.iotdb.db.exception.metadata.IllegalParameterOfPathException;
2627

2728
import org.slf4j.Logger;
2829
import org.slf4j.LoggerFactory;
2930

31+
import java.util.HashSet;
3032
import java.util.List;
3133
import java.util.Locale;
3234
import java.util.Map;
35+
import java.util.Set;
3336

3437
import static org.apache.iotdb.commons.conf.IoTDBConstant.LOSS;
3538
import static org.apache.iotdb.commons.conf.IoTDBConstant.SDT;
@@ -101,6 +104,26 @@ public static void checkSchemaMeasurementNames(List<String> measurements)
101104
}
102105
}
103106

107+
/** Validate the shared measurement/alias namespace before creating any member of a batch. */
108+
public static void checkSchemaMeasurementAliases(
109+
final PartialPath devicePath, final List<String> measurements, final List<String> aliases)
110+
throws MetadataException {
111+
if (aliases == null) {
112+
return;
113+
}
114+
final Set<String> measurementNames = new HashSet<>(measurements);
115+
final Set<String> aliasNames = new HashSet<>();
116+
for (int i = 0; i < measurements.size(); i++) {
117+
final String alias = aliases.get(i);
118+
if (alias != null
119+
&& (!aliasNames.add(alias)
120+
|| (measurementNames.contains(alias) && !alias.equals(measurements.get(i))))) {
121+
throw new AliasAlreadyExistException(
122+
devicePath.concatAsMeasurementPath(measurements.get(i)).getFullPath(), alias);
123+
}
124+
}
125+
}
126+
104127
/** check whether the database name uses illegal characters */
105128
public static void checkDatabase(String database) throws IllegalPathException {
106129
if (!IoTDBConfig.STORAGE_GROUP_PATTERN.matcher(database).matches()) {

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionAliasAndTagTest.java‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,10 @@
2222
import org.apache.iotdb.commons.exception.MetadataException;
2323
import org.apache.iotdb.commons.path.PartialPath;
2424
import org.apache.iotdb.commons.path.PathPatternTree;
25+
import org.apache.iotdb.db.exception.metadata.AliasAlreadyExistException;
2526
import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
2627
import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.ITimeSeriesSchemaInfo;
28+
import org.apache.iotdb.db.schemaengine.schemaregion.write.req.impl.CreateAlignedTimeSeriesPlanImpl;
2729

2830
import org.apache.tsfile.enums.TSDataType;
2931
import org.apache.tsfile.file.metadata.enums.CompressionType;
@@ -34,6 +36,7 @@
3436
import org.slf4j.Logger;
3537
import org.slf4j.LoggerFactory;
3638

39+
import java.util.ArrayList;
3740
import java.util.Arrays;
3841
import java.util.Collections;
3942
import java.util.HashMap;
@@ -58,6 +61,65 @@ public void setUp() throws Exception {
5861
schemaRegion = getSchemaRegion("root.sg", 0);
5962
}
6063

64+
@Test
65+
public void testAlignedBatchRejectsConflictingAliasesBeforeCreation() throws Exception {
66+
for (final List<String> aliases :
67+
Arrays.asList(
68+
Arrays.asList("dup", "dup"), Arrays.asList("s2", null), Arrays.asList(null, "s1"))) {
69+
final CreateAlignedTimeSeriesPlanImpl plan = alignedAliasPlan(aliases);
70+
try {
71+
schemaRegion.createAlignedTimeSeries(plan);
72+
Assert.fail("Expected alias collision");
73+
} catch (final AliasAlreadyExistException expected) {
74+
Assert.assertTrue(
75+
SchemaRegionTestUtil.showTimeseries(schemaRegion, new PartialPath("root.sg.batch.*"))
76+
.isEmpty());
77+
}
78+
}
79+
schemaRegion.createAlignedTimeSeries(alignedAliasPlan(Arrays.asList("a1", "a2")));
80+
final CreateAlignedTimeSeriesPlanImpl merge = alignedAliasPlan(Arrays.asList("dup", "dup"));
81+
merge.setWithMerge(true);
82+
try {
83+
schemaRegion.createAlignedTimeSeries(merge);
84+
Assert.fail("Expected alias collision during merge");
85+
} catch (final AliasAlreadyExistException expected) {
86+
checkAliasAndTagsAndAttributes(
87+
"root.sg.batch.s1", "a1", Collections.emptyMap(), Collections.emptyMap());
88+
checkAliasAndTagsAndAttributes(
89+
"root.sg.batch.s2", "a2", Collections.emptyMap(), Collections.emptyMap());
90+
}
91+
}
92+
93+
@Test
94+
public void testAlignedSelfAliasAndRemainingAliasAfterDeletion() throws Exception {
95+
schemaRegion.createAlignedTimeSeries(alignedAliasPlan(Arrays.asList("s1", "a2")));
96+
final PathPatternTree delete = new PathPatternTree();
97+
delete.appendPathPattern(new PartialPath("root.sg.batch.s2"));
98+
delete.constructTree();
99+
schemaRegion.constructSchemaBlackList(delete);
100+
schemaRegion.deleteTimeseriesInBlackList(delete);
101+
checkAliasAndTagsAndAttributes(
102+
"root.sg.batch.s1", "s1", Collections.emptyMap(), Collections.emptyMap());
103+
Assert.assertEquals(
104+
"root.sg.batch.s1",
105+
SchemaRegionTestUtil.showTimeseries(schemaRegion, new PartialPath("root.sg.batch.s1"))
106+
.get(0)
107+
.getFullPath());
108+
}
109+
110+
private CreateAlignedTimeSeriesPlanImpl alignedAliasPlan(final List<String> aliases)
111+
throws MetadataException {
112+
return new CreateAlignedTimeSeriesPlanImpl(
113+
new PartialPath("root.sg.batch"),
114+
new ArrayList<>(Arrays.asList("s1", "s2")),
115+
new ArrayList<>(Collections.nCopies(2, TSDataType.INT32)),
116+
new ArrayList<>(Collections.nCopies(2, TSEncoding.PLAIN)),
117+
new ArrayList<>(Collections.nCopies(2, CompressionType.SNAPPY)),
118+
new ArrayList<>(aliases),
119+
null,
120+
null);
121+
}
122+
61123
/**
62124
* Prepare timeseries
63125
*

0 commit comments

Comments
 (0)