diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/AbstractUpdateLastCacheOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/AbstractUpdateLastCacheOperator.java index e1c3cb335b095..38c5781d107fa 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/AbstractUpdateLastCacheOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/AbstractUpdateLastCacheOperator.java @@ -129,7 +129,6 @@ protected void mayUpdateLastCache( ? new TimeValuePair(time, value) : needUpdateNullEntry ? TableDeviceLastCache.PLACEHOLDER_EMPTY_COLUMN : null }, - fullPath.isUnderAlignedEntity(), new IMeasurementSchema[] {fullPath.getMeasurementSchema()}); return; } @@ -149,7 +148,6 @@ protected void mayUpdateLastCache( fullPath.getIDeviceID(), new String[] {fullPath.getMeasurement()}, new TimeValuePair[] {seriesScanInfo.right}, - fullPath.isUnderAlignedEntity(), new IMeasurementSchema[] {fullPath.getMeasurementSchema()}); } } finally { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java index ecf76892e4cd9..d053cfd1ec6d0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java @@ -927,7 +927,7 @@ private boolean canComposeTimeValuePair(final int columnIndex) { public void updateLastCache(String databaseName) { TreeDeviceSchemaCacheManager.getInstance() .updateLastCacheIfExists( - databaseName, getDeviceID(), measurements, this, isAligned, measurementSchemas); + databaseName, getDeviceID(), measurements, this, measurementSchemas); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java index 5c91d0af4eaaf..61c9a69536743 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java @@ -1405,12 +1405,7 @@ public void updateLastCache(final String databaseName, final TSStatus[] results) } TreeDeviceSchemaCacheManager.getInstance() .updateLastCacheIfExists( - databaseName, - getDeviceID(), - measurements, - timeValuePairs, - isAligned, - measurementSchemas); + databaseName, getDeviceID(), measurements, timeValuePairs, measurementSchemas); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java index c7f67307a7019..31c2054d6d24f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java @@ -131,33 +131,6 @@ null, new TreeDeviceTemplateSchema(database, deviceSchemaInfo.getTemplateId())) } } - int setMeasurementSchema( - final String database, - final boolean isAligned, - final String[] measurements, - final IMeasurementSchema[] schemas) { - if (schemas == null) { - return 0; - } - // Safe here because tree schema is invalidated by the whole entry - IDeviceSchema schema = deviceSchema.get(); - int result = 0; - if (schema == null) { - final TreeDeviceNormalSchema newSchema = new TreeDeviceNormalSchema(database, isAligned); - if (deviceSchema.compareAndSet(null, newSchema)) { - schema = newSchema; - result = TreeDeviceNormalSchema.INSTANCE_SIZE; - } else { - schema = deviceSchema.get(); - } - } - if (!(schema instanceof TreeDeviceNormalSchema)) { - return 0; - } - result += ((TreeDeviceNormalSchema) schema).update(measurements, schemas); - return deviceSchema.get() == schema ? result : 0; - } - IDeviceSchema getDeviceSchema() { return deviceSchema.get(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java index 17f691d85426a..b9b8881dc9c5e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java @@ -493,35 +493,30 @@ public IDeviceSchema getDeviceSchema(final IDeviceID deviceID) { return Objects.nonNull(entry) ? entry.getDeviceSchema() : null; } + /** + * Update tree-model last values without populating schema. Query plans and storage-engine writes + * may carry schema that has already been deleted; only schema fetches may populate that cache. + */ void updateLastCache( final String database, final IDeviceID deviceID, final String[] measurements, final @Nullable TimeValuePair[] timeValuePairs, - final boolean isAligned, final IMeasurementSchema[] measurementSchemas, final boolean initOrInvalidate) { - final String previousDatabase = treeModelDatabasePool.putIfAbsent(database, database); - final String database2Use = Objects.nonNull(previousDatabase) ? previousDatabase : database; - dualKeyCache.update( new TableId(null, deviceID.getTableName()), deviceID, Objects.isNull(timeValuePairs) ? new TableDeviceCacheEntry() : null, initOrInvalidate ? entry -> - entry.setMeasurementSchema( - database2Use, isAligned, measurements, measurementSchemas) - + entry.initOrInvalidateLastCache( - database, - deviceID.getTableName(), - measurements, - Objects.nonNull(timeValuePairs), - false) - : entry -> - entry.setMeasurementSchema( - database2Use, isAligned, measurements, measurementSchemas) - + entry.tryUpdateLastCache(measurements, measurementSchemas, timeValuePairs), + entry.initOrInvalidateLastCache( + database, + deviceID.getTableName(), + measurements, + Objects.nonNull(timeValuePairs), + false) + : entry -> entry.tryUpdateLastCache(measurements, measurementSchemas, timeValuePairs), Objects.isNull(timeValuePairs)); } @@ -530,18 +525,12 @@ void updateLastCache( final IDeviceID deviceID, final String[] measurements, final LastCacheUpdateSource updateSource, - final boolean isAligned, final IMeasurementSchema[] measurementSchemas) { - final String previousDatabase = treeModelDatabasePool.putIfAbsent(database, database); - final String database2Use = Objects.nonNull(previousDatabase) ? previousDatabase : database; - dualKeyCache.update( new TableId(null, deviceID.getTableName()), deviceID, null, - entry -> - entry.setMeasurementSchema(database2Use, isAligned, measurements, measurementSchemas) - + entry.tryUpdateLastCache(measurements, measurementSchemas, updateSource), + entry -> entry.tryUpdateLastCache(measurements, measurementSchemas, updateSource), false); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceSchemaCacheManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceSchemaCacheManager.java index 0f0f92ef0fffe..3746f048af90e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceSchemaCacheManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceSchemaCacheManager.java @@ -350,14 +350,13 @@ public void updateLastCacheIfExists( final IDeviceID deviceID, final String[] measurements, final @Nonnull TimeValuePair[] timeValuePairs, - final boolean isAligned, final IMeasurementSchema[] measurementSchemas) { if (!ClusterPartitionFetcher.getInstance().needLastCache(database)) { return; } tableDeviceSchemaCache.updateLastCache( - database, deviceID, measurements, timeValuePairs, isAligned, measurementSchemas, false); + database, deviceID, measurements, timeValuePairs, measurementSchemas, false); } public void updateLastCacheIfExists( @@ -365,10 +364,9 @@ public void updateLastCacheIfExists( final IDeviceID deviceID, final String[] measurements, final LastCacheUpdateSource updateSource, - final boolean isAligned, final IMeasurementSchema[] measurementSchemas) { tableDeviceSchemaCache.updateLastCache( - database, deviceID, measurements, updateSource, isAligned, measurementSchemas); + database, deviceID, measurements, updateSource, measurementSchemas); } /** @@ -382,9 +380,9 @@ public void updateLastCacheIfExists( * is written, but it won't affect the eventual consistency. * *

- Second time put the calculated {@link TimeValuePair}, and use {@link - * #updateLastCacheIfExists(String, IDeviceID, String[], TimeValuePair[], boolean, - * IMeasurementSchema[])}. The input {@link TimeValuePair} shall never be or contain {@code null}, - * if the measurement is with all {@code null}s, its {@link TimeValuePair} shall be {@link + * #updateLastCacheIfExists(String, IDeviceID, String[], TimeValuePair[], IMeasurementSchema[])}. + * The input {@link TimeValuePair} shall never be or contain {@code null}, if the measurement is + * with all {@code null}s, its {@link TimeValuePair} shall be {@link * TableDeviceLastCache#PLACEHOLDER_EMPTY_COLUMN}. This method is not supposed to update time * column. * @@ -401,8 +399,7 @@ public void declareLastCache(final String database, final MeasurementPath measur measurementPath.getIDeviceID(), new String[] {measurementPath.getMeasurement()}, null, - measurementPath.isUnderAlignedEntity(), - new IMeasurementSchema[] {measurementPath.getMeasurementSchema()}, + null, true); } @@ -421,8 +418,7 @@ public void invalidateLastCache(final String database, final MeasurementPath mea measurementPath.getIDeviceID(), new String[] {measurementPath.getMeasurement()}, new TimeValuePair[] {null}, - measurementPath.isUnderAlignedEntity(), - new IMeasurementSchema[] {measurementPath.getMeasurementSchema()}, + null, true); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index ee877628cd992..46c9339a507f0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -4745,10 +4745,8 @@ private void updateLastCache(TsFileResource newTsFileResource, TsFileLastReader TableDeviceSchemaCache.getInstance() .updateLastCacheIfExists(databaseName, deviceID, measurements, timeValuePairs); } else { - // we do not update schema here, so aligned is not relevant TreeDeviceSchemaCacheManager.getInstance() - .updateLastCacheIfExists( - databaseName, deviceID, measurements, timeValuePairs, false, null); + .updateLastCacheIfExists(databaseName, deviceID, measurements, timeValuePairs, null); } } newTsFileResource.setLastValues(null); @@ -4766,10 +4764,8 @@ private void updateLastCache(TsFileResource newTsFileResource, TsFileLastReader TableDeviceSchemaCache.getInstance() .updateLastCacheIfExists(databaseName, deviceID, measurements, timeValuePairs); } else { - // we do not update schema here, so aligned is not relevant TreeDeviceSchemaCacheManager.getInstance() - .updateLastCacheIfExists( - databaseName, deviceID, measurements, timeValuePairs, false, null); + .updateLastCacheIfExists(databaseName, deviceID, measurements, timeValuePairs, null); } } } else { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/cache/TreeDeviceSchemaCacheManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/cache/TreeDeviceSchemaCacheManagerTest.java index 59cdf62056126..7d7192ddf3e7a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/cache/TreeDeviceSchemaCacheManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/cache/TreeDeviceSchemaCacheManagerTest.java @@ -217,7 +217,6 @@ public void testUpdateLastCache() throws IllegalPathException { new TimeValuePair[] { new TimeValuePair(2, new TsPrimitiveType.TsInt(2)), }, - false, new MeasurementSchema[] {s1}); treeDeviceSchemaCacheManager.invalidateLastCache( database, new MeasurementPath(device.concatNode("s1"), s1)); @@ -229,7 +228,6 @@ public void testUpdateLastCache() throws IllegalPathException { StringArrayDeviceID.splitDeviceIdString(device.getNodes())), new String[] {"s2"}, new TimeValuePair[] {tv1}, - false, new MeasurementSchema[] {s2}); treeDeviceSchemaCacheManager.updateLastCacheIfExists( @@ -238,7 +236,6 @@ public void testUpdateLastCache() throws IllegalPathException { StringArrayDeviceID.splitDeviceIdString(device.getNodes())), new String[] {"s3"}, new TimeValuePair[] {tv1}, - false, new MeasurementSchema[] {s3}); Assert.assertNull( @@ -260,7 +257,6 @@ public void testUpdateLastCache() throws IllegalPathException { new TimeValuePair(2, new TsPrimitiveType.TsInt(2)), new TimeValuePair(2, new TsPrimitiveType.TsInt(2)) }, - false, measurementSchemas); Assert.assertNull( @@ -277,6 +273,91 @@ public void testUpdateLastCache() throws IllegalPathException { Assert.assertEquals(0, TableDeviceSchemaCache.getInstance().getMemoryUsage()); } + @Test + public void testLastCacheDoesNotRestoreDeletedSchema() throws IllegalPathException { + final String database = "root.db"; + for (boolean aligned : new boolean[] {false, true}) { + final PartialPath device = new PartialPath("root.db.d" + aligned); + final MeasurementSchema schema = new MeasurementSchema("s1", TSDataType.INT32); + final MeasurementPath path = new MeasurementPath(device.concatNode("s1"), schema, aligned); + final ClusterSchemaTree schemaTree = new ClusterSchemaTree(); + schemaTree.appendSingleMeasurementPath(path); + schemaTree.setDatabases(Collections.singleton(database)); + treeDeviceSchemaCacheManager.put(schemaTree); + Assert.assertFalse(treeDeviceSchemaCacheManager.getMatchedNormalSchema(path).isEmpty()); + + treeDeviceSchemaCacheManager.invalidate(Collections.singletonList(path)); + Assert.assertTrue(treeDeviceSchemaCacheManager.getMatchedNormalSchema(path).isEmpty()); + treeDeviceSchemaCacheManager.declareLastCache(database, path); + Assert.assertTrue( + "LAST initialization must not restore deleted schema", + treeDeviceSchemaCacheManager.getMatchedNormalSchema(path).isEmpty()); + + final TimeValuePair value = new TimeValuePair(1, new TsPrimitiveType.TsInt(1)); + treeDeviceSchemaCacheManager.updateLastCacheIfExists( + database, + path.getIDeviceID(), + new String[] {"s1"}, + new TimeValuePair[] {value}, + new MeasurementSchema[] {schema}); + Assert.assertEquals(value, treeDeviceSchemaCacheManager.getLastCache(path)); + Assert.assertTrue(treeDeviceSchemaCacheManager.getMatchedNormalSchema(path).isEmpty()); + Assert.assertTrue(treeDeviceSchemaCacheManager.get(device, new String[] {"s1"}).isEmpty()); + + treeDeviceSchemaCacheManager.invalidateLastCache(database, path); + Assert.assertNull(treeDeviceSchemaCacheManager.getLastCache(path)); + Assert.assertTrue(treeDeviceSchemaCacheManager.getMatchedNormalSchema(path).isEmpty()); + } + } + + @Test + public void testLastCacheUpdateDoesNotRestoreDeletedSchema() throws IllegalPathException { + final String database = "root.db"; + for (boolean aligned : new boolean[] {false, true}) { + final PartialPath device = new PartialPath("root.db.d" + aligned); + final MeasurementSchema deletedSchema = new MeasurementSchema("s1", TSDataType.INT32); + final MeasurementPath deletedPath = + new MeasurementPath(device.concatNode("s1"), deletedSchema, aligned); + final MeasurementPath remainingPath = + new MeasurementPath( + device.concatNode("s2"), new MeasurementSchema("s2", TSDataType.INT64), aligned); + final ClusterSchemaTree schemaTree = new ClusterSchemaTree(); + schemaTree.appendSingleMeasurementPath(deletedPath); + schemaTree.appendSingleMeasurementPath(remainingPath); + schemaTree.setDatabases(Collections.singleton(database)); + treeDeviceSchemaCacheManager.put(schemaTree); + treeDeviceSchemaCacheManager.declareLastCache(database, deletedPath); + + treeDeviceSchemaCacheManager.invalidate(Collections.singletonList(deletedPath)); + final ClusterSchemaTree remainingSchema = new ClusterSchemaTree(); + remainingSchema.appendSingleMeasurementPath(remainingPath); + remainingSchema.setDatabases(Collections.singleton(database)); + treeDeviceSchemaCacheManager.put(remainingSchema); + + treeDeviceSchemaCacheManager.updateLastCacheIfExists( + database, + deletedPath.getIDeviceID(), + new String[] {"s1"}, + new TimeValuePair[] {new TimeValuePair(1, new TsPrimitiveType.TsInt(1))}, + new MeasurementSchema[] {deletedSchema}); + Assert.assertTrue( + "LAST completion must not restore deleted schema in a recreated device entry", + treeDeviceSchemaCacheManager.getMatchedNormalSchema(deletedPath).isEmpty()); + Assert.assertNull(treeDeviceSchemaCacheManager.getLastCache(deletedPath)); + + treeDeviceSchemaCacheManager.invalidateLastCache(database, deletedPath); + Assert.assertTrue(treeDeviceSchemaCacheManager.getMatchedNormalSchema(deletedPath).isEmpty()); + Assert.assertEquals( + remainingPath.getMeasurementSchema(), + treeDeviceSchemaCacheManager + .getMatchedNormalSchema(remainingPath) + .searchMeasurementPaths(remainingPath) + .left + .get(0) + .getMeasurementSchema()); + } + } + @Test public void testLastCacheIsInvalidatedWhenDatabaseNeedLastCacheSetFalse() throws IllegalPathException { @@ -293,7 +374,6 @@ public void testLastCacheIsInvalidatedWhenDatabaseNeedLastCacheSetFalse() StringArrayDeviceID.splitDeviceIdString(device.getNodes())), new String[] {"s1"}, new TimeValuePair[] {tv1}, - false, new MeasurementSchema[] {s1}); Assert.assertEquals(tv1, treeDeviceSchemaCacheManager.getLastCache(measurementPath)); @@ -341,6 +421,7 @@ public TimeValuePair composeTimeValuePair(final int columnIndex) { insertRowNode.updateLastCache(database); Assert.assertEquals(1, composedValueCount.get()); + Assert.assertTrue(treeDeviceSchemaCacheManager.getMatchedNormalSchema(s1Path).isEmpty()); Assert.assertEquals( new TimeValuePair(1L, new TsPrimitiveType.TsInt(1)), treeDeviceSchemaCacheManager.getLastCache( @@ -398,7 +479,6 @@ private void updateLastCache( StringArrayDeviceID.splitDeviceIdString(devicePath.getNodes())), new String[] {path.getMeasurement()}, new TimeValuePair[] {timeValuePair}, - false, new MeasurementSchema[] {schema}); }