From 76990829096a6843d7221e1db7aa17147e7e8885 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 14 Oct 2026 14:34:34 +0800 Subject: [PATCH] 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 (cherry picked from commit a498d7d2a4608993c5c7d0c11f954e77b0554900) --- .../impl/mem/MTreeBelowSGMemoryImpl.java | 23 +--- .../mem/mnode/impl/BasicInternalMNode.java | 60 +++++--- .../impl/pbtree/MTreeBelowSGCachedImpl.java | 1 + .../schemaregion/utils/MetaFormatUtils.java | 23 ++++ .../SchemaRegionAliasAndTagTest.java | 62 +++++++++ .../impl/mem/MTreeBelowSGMemoryImplTest.java | 128 ++++++++++++++++++ 6 files changed, 261 insertions(+), 36 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java index a82250e229986..4f5ba04fdc59e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java @@ -45,6 +45,7 @@ import org.apache.iotdb.db.schemaengine.metric.SchemaRegionMemMetric; import org.apache.iotdb.db.schemaengine.rescon.MemSchemaRegionStatistics; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.IMemMNode; +import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.impl.BasicInternalMNode; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.loader.MNodeFactoryLoader; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.traverser.collector.EntityCollector; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.traverser.collector.MNodeCollector; @@ -374,6 +375,7 @@ public List> createAlignedTimeSeries( throws MetadataException { final List> measurementMNodeList = new ArrayList<>(); MetaFormatUtils.checkSchemaMeasurementNames(measurements); + MetaFormatUtils.checkSchemaMeasurementAliases(devicePath, measurements, aliasList); final IMemMNode deviceParent = checkAndAutoCreateInternalPath(devicePath); // synchronize check and add, we need addChild operation be atomic. @@ -607,28 +609,13 @@ public IMeasurementMNode deleteTimeseries(PartialPath path) throws Me public void deleteEmptyInternalMNode(IDeviceMNode entityMNode) { IMemMNode curNode = entityMNode.getAsMNode(); if (!entityMNode.isUseTemplate()) { - boolean hasMeasurement = false; - boolean hasNonViewMeasurement = false; - IMemMNode child; - try (final IMNodeIterator iterator = store.getChildrenIterator(curNode)) { - while (iterator.hasNext()) { - child = iterator.next(); - if (child.isMeasurement()) { - hasMeasurement = true; - if (!child.getAsMeasurementMNode().isLogicalView()) { - hasNonViewMeasurement = true; - break; - } - } - } - } - - if (!hasMeasurement) { + final BasicInternalMNode internalNode = (BasicInternalMNode) curNode.getBasicMNode(); + if (internalNode.getMeasurementChildCount() == 0) { synchronized (this) { curNode = store.setToInternal(entityMNode); } refreshAncestorsHavingDeviceDescendant(curNode.getParent()); - } else if (!hasNonViewMeasurement) { + } else if (internalNode.getNonViewMeasurementChildCount() == 0) { // has some measurement but they are all logical view entityMNode.setAligned(null); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/BasicInternalMNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/BasicInternalMNode.java index 904a0a5b5ba18..5786195be95c0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/BasicInternalMNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/BasicInternalMNode.java @@ -46,6 +46,10 @@ public class BasicInternalMNode extends BasicMNode implements IInternalMNode deviceInfo = null; + // Direct children only. Snapshot loading rebuilds these counters through addChild. + private int measurementChildCount; + private int nonViewMeasurementChildCount; + /** Constructor of MNode. */ public BasicInternalMNode(IMemMNode parent, String name) { super(parent, name); @@ -97,6 +101,9 @@ public IMemMNode addChild(String name, IMemMNode child) { } child.setParent(this); IMemMNode existingChild = children.putIfAbsent(name, child); + if (existingChild == null) { + updateMeasurementChildCount(child, 1); + } return existingChild == null ? child : existingChild; } @@ -113,21 +120,7 @@ public IMemMNode addChild(String name, IMemMNode child) { */ @Override public IMemMNode addChild(IMemMNode child) { - /* use cpu time to exchange memory - * measurementNode's children should be null to save memory - * add child method will only be called when writing MTree, which is not a frequent operation - */ - if (children == null) { - // double check, children is volatile - synchronized (this) { - if (children == null) { - children = new MemMNodeContainer(); - } - } - } - - child.setParent(this); - children.putIfAbsent(child.getName(), child); + addChild(child.getName(), child); return child; } @@ -135,7 +128,11 @@ public IMemMNode addChild(IMemMNode child) { @Override public IMemMNode deleteChild(String name) { if (children != null) { - return children.remove(name); + final IMemMNode child = children.remove(name); + if (child != null) { + updateMeasurementChildCount(child, -1); + } + return child; } return null; } @@ -151,12 +148,39 @@ public IMNodeContainer getChildren() { @Override public void setChildren(IMNodeContainer children) { this.children = children; + measurementChildCount = 0; + nonViewMeasurementChildCount = 0; + if (children != null) { + children.values().forEach(child -> updateMeasurementChildCount(child, 1)); + } + } + + private void updateMeasurementChildCount(final IMemMNode child, final int delta) { + if (child.isMeasurement()) { + measurementChildCount += delta; + if (!child.getAsMeasurementMNode().isLogicalView()) { + nonViewMeasurementChildCount += delta; + } + } + } + + public int getMeasurementChildCount() { + return measurementChildCount; + } + + public int getNonViewMeasurementChildCount() { + return nonViewMeasurementChildCount; } - /** MNodeContainer reference and basic occupation, 8 + 80B. DeviceInfo reference and size. */ + /** MNodeContainer reference and basic occupation, 8 + 80B. Counters, 8B. DeviceInfo and size. */ @Override public int estimateSize() { - return 8 + 80 + super.estimateSize() + 8 + (deviceInfo == null ? 0 : deviceInfo.estimateSize()); + return 8 + + 80 + + 8 + + super.estimateSize() + + 8 + + (deviceInfo == null ? 0 : deviceInfo.estimateSize()); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java index a326a9512d9bd..0b86ad4b7dc39 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java @@ -465,6 +465,7 @@ public List> createAlignedTimeSeries( throws MetadataException { final List> measurementMNodeList = new ArrayList<>(); MetaFormatUtils.checkSchemaMeasurementNames(measurements); + MetaFormatUtils.checkSchemaMeasurementAliases(devicePath, measurements, aliasList); final ICachedMNode deviceParent = checkAndAutoCreateInternalPath(devicePath); try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/MetaFormatUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/MetaFormatUtils.java index 1b7017e694055..239a2e0bd4fce 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/MetaFormatUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/MetaFormatUtils.java @@ -22,14 +22,17 @@ import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.conf.IoTDBConfig; +import org.apache.iotdb.db.exception.metadata.AliasAlreadyExistException; import org.apache.iotdb.db.exception.metadata.IllegalParameterOfPathException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.HashSet; import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Set; import static org.apache.iotdb.commons.conf.IoTDBConstant.LOSS; import static org.apache.iotdb.commons.conf.IoTDBConstant.SDT; @@ -101,6 +104,26 @@ public static void checkSchemaMeasurementNames(List measurements) } } + /** Validate the shared measurement/alias namespace before creating any member of a batch. */ + public static void checkSchemaMeasurementAliases( + final PartialPath devicePath, final List measurements, final List aliases) + throws MetadataException { + if (aliases == null) { + return; + } + final Set measurementNames = new HashSet<>(measurements); + final Set aliasNames = new HashSet<>(); + for (int i = 0; i < measurements.size(); i++) { + final String alias = aliases.get(i); + if (alias != null + && (!aliasNames.add(alias) + || (measurementNames.contains(alias) && !alias.equals(measurements.get(i))))) { + throw new AliasAlreadyExistException( + devicePath.concatAsMeasurementPath(measurements.get(i)).getFullPath(), alias); + } + } + } + /** check whether the database name uses illegal characters */ public static void checkDatabase(String database) throws IllegalPathException { if (!IoTDBConfig.STORAGE_GROUP_PATTERN.matcher(database).matches()) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionAliasAndTagTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionAliasAndTagTest.java index c6e283e11b6d2..96e43a15bb8bc 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionAliasAndTagTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionAliasAndTagTest.java @@ -22,8 +22,10 @@ import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.path.PathPatternTree; +import org.apache.iotdb.db.exception.metadata.AliasAlreadyExistException; import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion; import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.ITimeSeriesSchemaInfo; +import org.apache.iotdb.db.schemaengine.schemaregion.write.req.impl.CreateAlignedTimeSeriesPlanImpl; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.enums.CompressionType; @@ -34,6 +36,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -58,6 +61,65 @@ public void setUp() throws Exception { schemaRegion = getSchemaRegion("root.sg", 0); } + @Test + public void testAlignedBatchRejectsConflictingAliasesBeforeCreation() throws Exception { + for (final List aliases : + Arrays.asList( + Arrays.asList("dup", "dup"), Arrays.asList("s2", null), Arrays.asList(null, "s1"))) { + final CreateAlignedTimeSeriesPlanImpl plan = alignedAliasPlan(aliases); + try { + schemaRegion.createAlignedTimeSeries(plan); + Assert.fail("Expected alias collision"); + } catch (final AliasAlreadyExistException expected) { + Assert.assertTrue( + SchemaRegionTestUtil.showTimeseries(schemaRegion, new PartialPath("root.sg.batch.*")) + .isEmpty()); + } + } + schemaRegion.createAlignedTimeSeries(alignedAliasPlan(Arrays.asList("a1", "a2"))); + final CreateAlignedTimeSeriesPlanImpl merge = alignedAliasPlan(Arrays.asList("dup", "dup")); + merge.setWithMerge(true); + try { + schemaRegion.createAlignedTimeSeries(merge); + Assert.fail("Expected alias collision during merge"); + } catch (final AliasAlreadyExistException expected) { + checkAliasAndTagsAndAttributes( + "root.sg.batch.s1", "a1", Collections.emptyMap(), Collections.emptyMap()); + checkAliasAndTagsAndAttributes( + "root.sg.batch.s2", "a2", Collections.emptyMap(), Collections.emptyMap()); + } + } + + @Test + public void testAlignedSelfAliasAndRemainingAliasAfterDeletion() throws Exception { + schemaRegion.createAlignedTimeSeries(alignedAliasPlan(Arrays.asList("s1", "a2"))); + final PathPatternTree delete = new PathPatternTree(); + delete.appendPathPattern(new PartialPath("root.sg.batch.s2")); + delete.constructTree(); + schemaRegion.constructSchemaBlackList(delete); + schemaRegion.deleteTimeseriesInBlackList(delete); + checkAliasAndTagsAndAttributes( + "root.sg.batch.s1", "s1", Collections.emptyMap(), Collections.emptyMap()); + Assert.assertEquals( + "root.sg.batch.s1", + SchemaRegionTestUtil.showTimeseries(schemaRegion, new PartialPath("root.sg.batch.s1")) + .get(0) + .getFullPath()); + } + + private CreateAlignedTimeSeriesPlanImpl alignedAliasPlan(final List aliases) + throws MetadataException { + return new CreateAlignedTimeSeriesPlanImpl( + new PartialPath("root.sg.batch"), + new ArrayList<>(Arrays.asList("s1", "s2")), + new ArrayList<>(Collections.nCopies(2, TSDataType.INT32)), + new ArrayList<>(Collections.nCopies(2, TSEncoding.PLAIN)), + new ArrayList<>(Collections.nCopies(2, CompressionType.SNAPPY)), + new ArrayList<>(aliases), + null, + null); + } + /** * Prepare timeseries * diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImplTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImplTest.java index 1dfa4a056bbb1..7f80bc184953a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImplTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImplTest.java @@ -31,6 +31,8 @@ import org.apache.iotdb.db.schemaengine.rescon.MemSchemaEngineStatistics; import org.apache.iotdb.db.schemaengine.rescon.MemSchemaRegionStatistics; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.IMemMNode; +import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.container.MemMNodeContainer; +import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.impl.BasicInternalMNode; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.loader.MNodeFactoryLoader; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.traverser.collector.MeasurementCollector; import org.apache.iotdb.db.schemaengine.template.Template; @@ -40,15 +42,141 @@ import org.apache.tsfile.file.metadata.enums.TSEncoding; import org.apache.tsfile.write.schema.MeasurementSchema; import org.junit.Assert; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import java.io.File; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; public class MTreeBelowSGMemoryImplTest { + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void testWideDeviceDeletionDoesNotScanChildrenPerMeasurement() throws Exception { + final MTreeBelowSGMemoryImpl mtree = newMTree(); + final int measurementCount = 1024; + final List measurements = new ArrayList<>(); + for (int i = 0; i < measurementCount; i++) { + measurements.add("s" + i); + } + mtree.createAlignedTimeSeries( + new PartialPath("root.sg.d"), + measurements, + Collections.nCopies(measurementCount, TSDataType.INT32), + Collections.nCopies(measurementCount, TSEncoding.PLAIN), + Collections.nCopies(measurementCount, CompressionType.SNAPPY), + null, + false, + Collections.emptySet(), + new AtomicBoolean(true)); + final IMemMNode device = mtree.getNodeByPath(new PartialPath("root.sg.d")); + final CountingChildren children = new CountingChildren(); + children.putAll(device.getChildren()); + device.setChildren(children); + final PartialPath pattern = new PartialPath("root.sg.d.*"); + mtree.constructSchemaBlackList(pattern, new AtomicBoolean()); + final List paths = mtree.getPreDeletedTimeseries(pattern); + Assert.assertEquals(measurementCount, paths.size()); + children.scans = 0; + for (final PartialPath path : paths) { + mtree.deleteTimeseries(path); + } + Assert.assertEquals(0, children.scans); + assertPathNotExist(mtree, new PartialPath("root.sg.d")); + Assert.assertEquals( + 0, mtree.getNodeByPath(new PartialPath("root.sg")).getSubtreeMeasurementCount()); + } + + @Test + public void testDirectChildCountsAfterSnapshotWithNestedDevice() throws Exception { + final MTreeBelowSGMemoryImpl mtree = newMTree(); + for (final String path : Arrays.asList("root.sg.d.s1", "root.sg.d.child.s2")) { + mtree.createTimeSeries( + new MeasurementPath(path), + TSDataType.INT32, + TSEncoding.PLAIN, + CompressionType.SNAPPY, + null, + null, + false, + null); + } + final File snapshotDir = temporaryFolder.newFolder(); + Assert.assertTrue(mtree.createSnapshot(snapshotDir)); + final MemSchemaRegionStatistics statistics = newRegionStatistics(); + final MTreeBelowSGMemoryImpl recovered = + MTreeBelowSGMemoryImpl.loadFromSnapshot( + snapshotDir, + "root.sg", + statistics, + new SchemaRegionMemMetric(statistics, "root.sg"), + node -> {}, + node -> {}, + node -> Collections.emptyMap(), + node -> Collections.emptyMap()); + final IMemMNode device = recovered.getNodeByPath(new PartialPath("root.sg.d")); + Assert.assertEquals( + 1, ((BasicInternalMNode) device.getBasicMNode()).getMeasurementChildCount()); + Assert.assertEquals(2, device.getSubtreeMeasurementCount()); + recovered.deleteTimeseries(new PartialPath("root.sg.d.s1")); + Assert.assertFalse(recovered.getNodeByPath(new PartialPath("root.sg.d")).isDevice()); + Assert.assertTrue(recovered.getNodeByPath(new PartialPath("root.sg.d.child")).isDevice()); + recovered.deleteTimeseries(new PartialPath("root.sg.d.child.s2")); + assertPathNotExist(recovered, new PartialPath("root.sg.d")); + } + + @Test + public void testAlignmentResetsWhenOnlyLogicalViewsRemain() throws Exception { + final MTreeBelowSGMemoryImpl mtree = newMTree(); + mtree.createTimeSeries( + new MeasurementPath("root.sg.d.s"), + TSDataType.INT32, + TSEncoding.PLAIN, + CompressionType.SNAPPY, + null, + null, + false, + null); + final IMemMNode device = mtree.getNodeByPath(new PartialPath("root.sg.d")); + // Model a logical view child to verify cleanup semantics independently of view creation. + final IMemMNode view = mock(IMemMNode.class); + final IMeasurementMNode viewMeasurement = mock(IMeasurementMNode.class); + when(view.isMeasurement()).thenReturn(true); + when(view.getAsMeasurementMNode()).thenReturn(viewMeasurement); + when(viewMeasurement.isLogicalView()).thenReturn(true); + device.addChild("view", view); + device.addChild("view", view); + Assert.assertEquals( + 2, ((BasicInternalMNode) device.getBasicMNode()).getMeasurementChildCount()); + mtree.deleteTimeseries(new PartialPath("root.sg.d.s")); + Assert.assertTrue(device.isDevice()); + Assert.assertNull(device.getAsDeviceMNode().isAlignedNullable()); + device.deleteChild("view"); + mtree.deleteEmptyInternalMNode(device.getAsDeviceMNode()); + assertPathNotExist(mtree, new PartialPath("root.sg.d")); + } + + private static class CountingChildren extends MemMNodeContainer { + private int scans; + + @Override + public Collection values() { + scans++; + return super.values(); + } + } + @Test public void testDeviceDescendantFlagIsMaintainedAcrossCreateDeleteAndRebuild() throws Exception { final MTreeBelowSGMemoryImpl mtree = newMTree();