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();