| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| package org.apache.iotdb.db.metadata.cache; |
| |
| import org.apache.iotdb.commons.exception.IllegalPathException; |
| import org.apache.iotdb.commons.path.MeasurementPath; |
| import org.apache.iotdb.commons.path.PartialPath; |
| import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree; |
| import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree; |
| import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache; |
| import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.SchemaCacheEntry; |
| import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager; |
| import org.apache.iotdb.db.schemaengine.template.Template; |
| import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; |
| import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; |
| import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; |
| import org.apache.iotdb.tsfile.read.TimeValuePair; |
| import org.apache.iotdb.tsfile.utils.TsPrimitiveType; |
| import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; |
| |
| import org.junit.After; |
| import org.junit.Assert; |
| import org.junit.Before; |
| import org.junit.Test; |
| |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.stream.Collectors; |
| |
| import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_MATCH_PATTERN; |
| |
| public class DataNodeSchemaCacheTest { |
| DataNodeSchemaCache dataNodeSchemaCache; |
| private Map<String, String> s1TagMap; |
| |
| @Before |
| public void setUp() throws Exception { |
| dataNodeSchemaCache = DataNodeSchemaCache.getInstance(); |
| s1TagMap = new HashMap<>(); |
| s1TagMap.put("k1", "v1"); |
| } |
| |
| @After |
| public void tearDown() throws Exception { |
| dataNodeSchemaCache.cleanUp(); |
| ClusterTemplateManager.getInstance().clear(); |
| } |
| |
| @Test |
| public void testGetSchemaEntity() throws IllegalPathException { |
| PartialPath device1 = new PartialPath("root.sg1.d1"); |
| String[] measurements = new String[3]; |
| measurements[0] = "s1"; |
| measurements[1] = "s2"; |
| measurements[2] = "s3"; |
| |
| dataNodeSchemaCache.put((ClusterSchemaTree) generateSchemaTree1()); |
| |
| Map<PartialPath, SchemaCacheEntry> schemaCacheEntryMap = |
| dataNodeSchemaCache.get(device1, measurements).getAllDevices().stream() |
| .flatMap(deviceSchemaInfo -> deviceSchemaInfo.getMeasurementSchemaPathList().stream()) |
| .collect( |
| Collectors.toMap( |
| o -> new PartialPath(o.getNodes()), |
| o -> |
| new SchemaCacheEntry( |
| "root.sg1", |
| o.getMeasurementSchema(), |
| o.getTagMap(), |
| o.isUnderAlignedEntity()))); |
| Assert.assertEquals( |
| TSDataType.INT32, |
| schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s1")).getTsDataType()); |
| Assert.assertEquals( |
| s1TagMap, schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s1")).getTagMap()); |
| Assert.assertEquals( |
| TSDataType.FLOAT, |
| schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s2")).getTsDataType()); |
| Assert.assertNull(schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s2")).getTagMap()); |
| Assert.assertEquals( |
| TSDataType.BOOLEAN, |
| schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s3")).getTsDataType()); |
| Assert.assertNull(schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s3")).getTagMap()); |
| |
| String[] otherMeasurements = new String[3]; |
| otherMeasurements[0] = "s3"; |
| otherMeasurements[1] = "s4"; |
| otherMeasurements[2] = "s5"; |
| |
| dataNodeSchemaCache.put((ClusterSchemaTree) generateSchemaTree2()); |
| |
| schemaCacheEntryMap = |
| dataNodeSchemaCache.get(device1, otherMeasurements).getAllDevices().stream() |
| .flatMap(deviceSchemaInfo -> deviceSchemaInfo.getMeasurementSchemaPathList().stream()) |
| .collect( |
| Collectors.toMap( |
| o -> new PartialPath(o.getNodes()), |
| o -> |
| new SchemaCacheEntry( |
| "root.sg1", |
| o.getMeasurementSchema(), |
| o.getTagMap(), |
| o.isUnderAlignedEntity()))); |
| Assert.assertEquals( |
| TSDataType.BOOLEAN, |
| schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s3")).getTsDataType()); |
| Assert.assertNull(schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s3")).getTagMap()); |
| Assert.assertEquals( |
| TSDataType.TEXT, |
| schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s4")).getTsDataType()); |
| Assert.assertNull(schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s4")).getTagMap()); |
| Assert.assertEquals( |
| TSDataType.INT64, |
| schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s5")).getTsDataType()); |
| Assert.assertNull(schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s4")).getTagMap()); |
| } |
| |
| @Test |
| public void testLastCache() throws IllegalPathException { |
| // test no cache |
| PartialPath devicePath = new PartialPath("root.sg1.d1"); |
| PartialPath seriesPath1 = new PartialPath("root.sg1.d1.s1"); |
| PartialPath seriesPath2 = new PartialPath("root.sg1.d1.s2"); |
| PartialPath seriesPath3 = new PartialPath("root.sg1.d1.s3"); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath1)); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath2)); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath3)); |
| // test no last cache |
| dataNodeSchemaCache.put((ClusterSchemaTree) generateSchemaTree1()); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath1)); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath2)); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath3)); |
| // put cache |
| long timestamp = 100; |
| long timestamp2 = 101; |
| TsPrimitiveType value = TsPrimitiveType.getByType(TSDataType.INT32, 101); |
| TsPrimitiveType value2 = TsPrimitiveType.getByType(TSDataType.INT32, 100); |
| TsPrimitiveType value3 = TsPrimitiveType.getByType(TSDataType.INT32, 99); |
| |
| // put into last cache when cache not exist |
| TimeValuePair timeValuePair = new TimeValuePair(timestamp, value); |
| dataNodeSchemaCache.updateLastCache(devicePath, "s1", timeValuePair, false, 99L); |
| TimeValuePair cachedTimeValuePair = dataNodeSchemaCache.getLastCache(seriesPath1); |
| Assert.assertNotNull(cachedTimeValuePair); |
| Assert.assertEquals(timestamp, cachedTimeValuePair.getTimestamp()); |
| Assert.assertEquals(value, cachedTimeValuePair.getValue()); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath2)); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath3)); |
| |
| // same time but low priority |
| TimeValuePair timeValuePair2 = new TimeValuePair(timestamp, value2); |
| dataNodeSchemaCache.updateLastCache(devicePath, "s1", timeValuePair2, false, 100L); |
| TimeValuePair cachedTimeValuePair2 = dataNodeSchemaCache.getLastCache(seriesPath1); |
| Assert.assertNotNull(cachedTimeValuePair2); |
| Assert.assertEquals(timestamp, cachedTimeValuePair2.getTimestamp()); |
| Assert.assertEquals(value, cachedTimeValuePair2.getValue()); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath2)); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath3)); |
| |
| // same time but high priority |
| dataNodeSchemaCache.updateLastCache(devicePath, "s1", timeValuePair2, true, 100L); |
| cachedTimeValuePair2 = dataNodeSchemaCache.getLastCache(seriesPath1); |
| Assert.assertNotNull(cachedTimeValuePair2); |
| Assert.assertEquals(timestamp, cachedTimeValuePair2.getTimestamp()); |
| Assert.assertEquals(value2, cachedTimeValuePair2.getValue()); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath2)); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath3)); |
| |
| // put into last cache when cache already exist |
| TimeValuePair timeValuePair3 = new TimeValuePair(timestamp2, value3); |
| dataNodeSchemaCache.updateLastCache(devicePath, "s1", timeValuePair3, false, 100L); |
| TimeValuePair cachedTimeValuePair3 = dataNodeSchemaCache.getLastCache(seriesPath1); |
| Assert.assertNotNull(cachedTimeValuePair3); |
| Assert.assertEquals(timestamp2, cachedTimeValuePair3.getTimestamp()); |
| Assert.assertEquals(value3, cachedTimeValuePair3.getValue()); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath2)); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(seriesPath3)); |
| } |
| |
| private ISchemaTree generateSchemaTree1() throws IllegalPathException { |
| ClusterSchemaTree schemaTree = new ClusterSchemaTree(); |
| Map<String, String> s1TagMap = new HashMap<>(); |
| s1TagMap.put("k1", "v1"); |
| schemaTree.appendSingleMeasurement( |
| new PartialPath("root.sg1.d1.s1"), |
| new MeasurementSchema("s1", TSDataType.INT32), |
| s1TagMap, |
| null, |
| false); |
| schemaTree.appendSingleMeasurement( |
| new PartialPath("root.sg1.d1.s2"), |
| new MeasurementSchema("s2", TSDataType.FLOAT), |
| null, |
| null, |
| false); |
| schemaTree.appendSingleMeasurement( |
| new PartialPath("root.sg1.d1.s3"), |
| new MeasurementSchema("s3", TSDataType.BOOLEAN), |
| null, |
| null, |
| false); |
| schemaTree.setDatabases(Collections.singleton("root.sg1")); |
| return schemaTree; |
| } |
| |
| private ISchemaTree generateSchemaTree2() throws IllegalPathException { |
| ClusterSchemaTree schemaTree = new ClusterSchemaTree(); |
| |
| schemaTree.appendSingleMeasurement( |
| new PartialPath("root.sg1.d1.s3"), |
| new MeasurementSchema("s3", TSDataType.BOOLEAN), |
| null, |
| null, |
| false); |
| schemaTree.appendSingleMeasurement( |
| new PartialPath("root.sg1.d1.s4"), |
| new MeasurementSchema("s4", TSDataType.TEXT), |
| null, |
| null, |
| false); |
| schemaTree.appendSingleMeasurement( |
| new PartialPath("root.sg1.d1.s5"), |
| new MeasurementSchema("s5", TSDataType.INT64), |
| null, |
| null, |
| false); |
| schemaTree.setDatabases(Collections.singleton("root.sg1")); |
| return schemaTree; |
| } |
| |
| @Test |
| public void testUpdateLastCache() throws IllegalPathException { |
| String database = "root.db"; |
| PartialPath device = new PartialPath("root.db.d"); |
| |
| String[] measurements = new String[] {"s1", "s2", "s3"}; |
| MeasurementSchema[] measurementSchemas = |
| new MeasurementSchema[] { |
| new MeasurementSchema("s1", TSDataType.INT32), |
| new MeasurementSchema("s2", TSDataType.INT32), |
| new MeasurementSchema("s3", TSDataType.INT32) |
| }; |
| |
| dataNodeSchemaCache.updateLastCache( |
| database, |
| device, |
| measurements, |
| measurementSchemas, |
| true, |
| index -> new TimeValuePair(1, new TsPrimitiveType.TsInt(1)), |
| index -> index != 1, |
| true, |
| 1L); |
| |
| Assert.assertNotNull(dataNodeSchemaCache.getLastCache(new PartialPath("root.db.d.s1"))); |
| Assert.assertNull(dataNodeSchemaCache.getLastCache(new PartialPath("root.db.d.s2"))); |
| Assert.assertNotNull(dataNodeSchemaCache.getLastCache(new PartialPath("root.db.d.s3"))); |
| |
| dataNodeSchemaCache.updateLastCache( |
| database, |
| device, |
| measurements, |
| measurementSchemas, |
| true, |
| index -> new TimeValuePair(2, new TsPrimitiveType.TsInt(2)), |
| index -> true, |
| true, |
| 1L); |
| |
| Assert.assertEquals( |
| new TimeValuePair(2, new TsPrimitiveType.TsInt(2)), |
| dataNodeSchemaCache.getLastCache(new PartialPath("root.db.d.s1"))); |
| Assert.assertEquals( |
| new TimeValuePair(2, new TsPrimitiveType.TsInt(2)), |
| dataNodeSchemaCache.getLastCache(new PartialPath("root.db.d.s2"))); |
| Assert.assertEquals( |
| new TimeValuePair(2, new TsPrimitiveType.TsInt(2)), |
| dataNodeSchemaCache.getLastCache(new PartialPath("root.db.d.s3"))); |
| } |
| |
| @Test |
| public void testPut() throws Exception { |
| ClusterSchemaTree clusterSchemaTree = new ClusterSchemaTree(); |
| Template template1 = |
| new Template( |
| "t1", |
| Arrays.asList("s1", "s2"), |
| Arrays.asList(TSDataType.DOUBLE, TSDataType.INT32), |
| Arrays.asList(TSEncoding.RLE, TSEncoding.RLE), |
| Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY)); |
| template1.setId(1); |
| Template template2 = |
| new Template( |
| "t2", |
| Arrays.asList("t1", "t2", "t3"), |
| Arrays.asList(TSDataType.DOUBLE, TSDataType.INT32, TSDataType.INT64), |
| Arrays.asList(TSEncoding.RLE, TSEncoding.RLE, TSEncoding.RLBE), |
| Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY, CompressionType.SNAPPY)); |
| template2.setId(2); |
| ClusterTemplateManager.getInstance().putTemplate(template1); |
| ClusterTemplateManager.getInstance().putTemplate(template2); |
| clusterSchemaTree.appendTemplateDevice(new PartialPath("root.sg1.d1"), false, 1, template1); |
| clusterSchemaTree.appendTemplateDevice(new PartialPath("root.sg1.d2"), false, 2, template2); |
| clusterSchemaTree.setDatabases(Collections.singleton("root.sg1")); |
| clusterSchemaTree.appendSingleMeasurementPath( |
| new MeasurementPath("root.sg1.d3.s1", TSDataType.FLOAT)); |
| dataNodeSchemaCache.put(clusterSchemaTree); |
| ClusterSchemaTree d1Tree = |
| dataNodeSchemaCache.getMatchedSchemaWithTemplate(new PartialPath("root.sg1.d1")); |
| ClusterSchemaTree d2Tree = |
| dataNodeSchemaCache.getMatchedSchemaWithTemplate(new PartialPath("root.sg1.d2")); |
| ClusterSchemaTree d3Tree = |
| dataNodeSchemaCache.getMatchedSchemaWithoutTemplate(new PartialPath("root.sg1.d3.s1")); |
| List<MeasurementPath> measurementPaths = d1Tree.searchMeasurementPaths(ALL_MATCH_PATTERN).left; |
| Assert.assertEquals(2, measurementPaths.size()); |
| for (MeasurementPath measurementPath : measurementPaths) { |
| Assert.assertEquals( |
| template1.getSchema(measurementPath.getMeasurement()), |
| measurementPath.getMeasurementSchema()); |
| } |
| measurementPaths = d2Tree.searchMeasurementPaths(ALL_MATCH_PATTERN).left; |
| Assert.assertEquals(3, measurementPaths.size()); |
| for (MeasurementPath measurementPath : measurementPaths) { |
| Assert.assertEquals( |
| template2.getSchema(measurementPath.getMeasurement()), |
| measurementPath.getMeasurementSchema()); |
| } |
| measurementPaths = d3Tree.searchMeasurementPaths(ALL_MATCH_PATTERN).left; |
| Assert.assertEquals(1, measurementPaths.size()); |
| Assert.assertEquals(TSDataType.FLOAT, measurementPaths.get(0).getMeasurementSchema().getType()); |
| Assert.assertEquals("root.sg1.d3.s1", measurementPaths.get(0).getFullPath()); |
| } |
| } |
