This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch rc/2.0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rc/2.0.11 by this push:
new ba3d598b9c7 [Performance] Reduce temporary allocations on insert path
(#18240)
ba3d598b9c7 is described below
commit ba3d598b9c7b72821bc86ec2dfd222a04bd77f76
Author: Caideyipi <[email protected]>
AuthorDate: Mon Aug 24 16:21:16 2026 +0800
[Performance] Reduce temporary allocations on insert path (#18240)
---
.../resource/memory/InsertNodeMemoryEstimator.java | 121 ++++++++++++++------
.../planner/plan/node/write/InsertRowNode.java | 29 +++--
.../plan/node/write/RelationalInsertRowNode.java | 7 +-
.../fetcher/cache/LastCacheUpdateSource.java | 37 ++++++
.../fetcher/cache/TableDeviceCacheEntry.java | 35 ++++--
.../fetcher/cache/TableDeviceLastCache.java | 124 ++++++++++++++-------
.../fetcher/cache/TableDeviceSchemaCache.java | 36 +++++-
.../cache/TreeDeviceSchemaCacheManager.java | 11 ++
8 files changed, 304 insertions(+), 96 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java
index 54655704c1c..49561cdba19 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java
@@ -155,6 +155,12 @@ public class InsertNodeMemoryEstimator {
// from the actual result because the properties of the parent class are not
added.
private static final double INSERT_ROW_NODE_EXPANSION_FACTOR = 1.3;
+ // Insert nodes are estimated on write threads. Reuse the identity set
between events,
+ // but discard unusually large sets to avoid retaining a large table on
every write thread.
+ private static final int MAX_RETAINED_DEDUPLICATED_OBJECTS = 1024;
+ private static final ThreadLocal<Set<Object>> REUSABLE_DEDUPLICATED_OBJECTS =
+
ThreadLocal.withInitial(InsertNodeMemoryEstimator::newDeduplicatedObjectSet);
+
public static long sizeOf(final InsertNode insertNode) {
try {
final String className = insertNode.getClass().getSimpleName();
@@ -220,7 +226,12 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfInsertTabletNode(final InsertTabletNode node) {
- return sizeOfInsertTabletNode(node, newDeduplicatedObjectSet());
+ final Set<Object> deduplicatedObjects = acquireDeduplicatedObjectSet();
+ try {
+ return sizeOfInsertTabletNode(node, deduplicatedObjects);
+ } finally {
+ releaseDeduplicatedObjectSet(deduplicatedObjects);
+ }
}
private static long sizeOfInsertTabletNode(
@@ -235,7 +246,12 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfInsertRowNode(final InsertRowNode node) {
- return sizeOfInsertRowNode(node, newDeduplicatedObjectSet());
+ final Set<Object> deduplicatedObjects = acquireDeduplicatedObjectSet();
+ try {
+ return sizeOfInsertRowNode(node, deduplicatedObjects);
+ } finally {
+ releaseDeduplicatedObjectSet(deduplicatedObjects);
+ }
}
private static long sizeOfInsertRowNode(
@@ -247,47 +263,68 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfInsertRowsNode(final InsertRowsNode node) {
- final Set<Object> deduplicatedObjects = newDeduplicatedObjectSet();
- long size = INSERT_ROWS_NODE_SIZE;
- size += calculateFullInsertNodeSize(node, deduplicatedObjects);
- size += sizeOfInsertRowNodeList(node.getInsertRowNodeList(),
deduplicatedObjects);
- size += sizeOfIntegerList(node.getInsertRowNodeIndexList());
- size += sizeOfResults(node.getResults());
- return size;
+ final Set<Object> deduplicatedObjects = acquireDeduplicatedObjectSet();
+ try {
+ long size = INSERT_ROWS_NODE_SIZE;
+ size += calculateFullInsertNodeSize(node, deduplicatedObjects);
+ size += sizeOfInsertRowNodeList(node.getInsertRowNodeList(),
deduplicatedObjects);
+ size += sizeOfIntegerList(node.getInsertRowNodeIndexList());
+ size += sizeOfResults(node.getResults());
+ return size;
+ } finally {
+ releaseDeduplicatedObjectSet(deduplicatedObjects);
+ }
}
private static long sizeOfInsertRowsOfOneDeviceNode(final
InsertRowsOfOneDeviceNode node) {
- final Set<Object> deduplicatedObjects = newDeduplicatedObjectSet();
- long size = INSERT_ROWS_OF_ONE_DEVICE_NODE_SIZE;
- size += calculateFullInsertNodeSize(node, deduplicatedObjects);
- size += sizeOfInsertRowNodeList(node.getInsertRowNodeList(),
deduplicatedObjects);
- size += sizeOfIntegerList(node.getInsertRowNodeIndexList());
- size += sizeOfResults(node.getResults());
- return size;
+ final Set<Object> deduplicatedObjects = acquireDeduplicatedObjectSet();
+ try {
+ long size = INSERT_ROWS_OF_ONE_DEVICE_NODE_SIZE;
+ size += calculateFullInsertNodeSize(node, deduplicatedObjects);
+ size += sizeOfInsertRowNodeList(node.getInsertRowNodeList(),
deduplicatedObjects);
+ size += sizeOfIntegerList(node.getInsertRowNodeIndexList());
+ size += sizeOfResults(node.getResults());
+ return size;
+ } finally {
+ releaseDeduplicatedObjectSet(deduplicatedObjects);
+ }
}
private static long sizeOfInsertMultiTabletsNode(final
InsertMultiTabletsNode node) {
- final Set<Object> deduplicatedObjects = newDeduplicatedObjectSet();
- long size = INSERT_MULTI_TABLETS_NODE_SIZE;
- size += calculateFullInsertNodeSize(node, deduplicatedObjects);
- size += sizeOfInsertTabletNodeList(node.getInsertTabletNodeList(),
deduplicatedObjects);
- size += sizeOfIntegerList(node.getParentInsertTabletNodeIndexList());
- size += sizeOfResults(node.getResults());
- return size;
+ final Set<Object> deduplicatedObjects = acquireDeduplicatedObjectSet();
+ try {
+ long size = INSERT_MULTI_TABLETS_NODE_SIZE;
+ size += calculateFullInsertNodeSize(node, deduplicatedObjects);
+ size += sizeOfInsertTabletNodeList(node.getInsertTabletNodeList(),
deduplicatedObjects);
+ size += sizeOfIntegerList(node.getParentInsertTabletNodeIndexList());
+ size += sizeOfResults(node.getResults());
+ return size;
+ } finally {
+ releaseDeduplicatedObjectSet(deduplicatedObjects);
+ }
}
private static long sizeOfRelationalInsertRowsNode(final
RelationalInsertRowsNode node) {
- final Set<Object> deduplicatedObjects = newDeduplicatedObjectSet();
- long size = RELATIONAL_INSERT_ROWS_NODE_SIZE;
- size += calculateFullInsertNodeSize(node, deduplicatedObjects);
- size += sizeOfInsertRowNodeList(node.getInsertRowNodeList(),
deduplicatedObjects);
- size += sizeOfIntegerList(node.getInsertRowNodeIndexList());
- // ignore deviceIDs
- return size;
+ final Set<Object> deduplicatedObjects = acquireDeduplicatedObjectSet();
+ try {
+ long size = RELATIONAL_INSERT_ROWS_NODE_SIZE;
+ size += calculateFullInsertNodeSize(node, deduplicatedObjects);
+ size += sizeOfInsertRowNodeList(node.getInsertRowNodeList(),
deduplicatedObjects);
+ size += sizeOfIntegerList(node.getInsertRowNodeIndexList());
+ // ignore deviceIDs
+ return size;
+ } finally {
+ releaseDeduplicatedObjectSet(deduplicatedObjects);
+ }
}
private static long sizeOfRelationalInsertRowNode(final
RelationalInsertRowNode node) {
- return sizeOfRelationalInsertRowNode(node, newDeduplicatedObjectSet());
+ final Set<Object> deduplicatedObjects = acquireDeduplicatedObjectSet();
+ try {
+ return sizeOfRelationalInsertRowNode(node, deduplicatedObjects);
+ } finally {
+ releaseDeduplicatedObjectSet(deduplicatedObjects);
+ }
}
private static long sizeOfRelationalInsertRowNode(
@@ -299,7 +336,12 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfRelationalInsertTabletNode(final
RelationalInsertTabletNode node) {
- return sizeOfRelationalInsertTabletNode(node, newDeduplicatedObjectSet());
+ final Set<Object> deduplicatedObjects = acquireDeduplicatedObjectSet();
+ try {
+ return sizeOfRelationalInsertTabletNode(node, deduplicatedObjects);
+ } finally {
+ releaseDeduplicatedObjectSet(deduplicatedObjects);
+ }
}
private static long sizeOfRelationalInsertTabletNode(
@@ -771,6 +813,21 @@ public class InsertNodeMemoryEstimator {
return Collections.newSetFromMap(new IdentityHashMap<>());
}
+ private static Set<Object> acquireDeduplicatedObjectSet() {
+ return REUSABLE_DEDUPLICATED_OBJECTS.get();
+ }
+
+ private static void releaseDeduplicatedObjectSet(final Set<Object>
deduplicatedObjects) {
+ if (deduplicatedObjects == null) {
+ return;
+ }
+ final boolean oversized = deduplicatedObjects.size() >
MAX_RETAINED_DEDUPLICATED_OBJECTS;
+ deduplicatedObjects.clear();
+ if (oversized) {
+ REUSABLE_DEDUPLICATED_OBJECTS.remove();
+ }
+ }
+
private static boolean shouldCountObject(
final Object object, final Set<Object> deduplicatedObjects) {
return object != null && (deduplicatedObjects == null ||
deduplicatedObjects.add(object));
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 04768c57b50..5a3628ac06c 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
@@ -33,6 +33,7 @@ import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode;
+import
org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.LastCacheUpdateSource;
import
org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TreeDeviceSchemaCacheManager;
import org.apache.iotdb.db.storageengine.dataregion.memtable.AbstractMemTable;
import
org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunkGroup;
@@ -60,7 +61,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Objects;
-public class InsertRowNode extends InsertNode implements WALEntryValue {
+public class InsertRowNode extends InsertNode implements WALEntryValue,
LastCacheUpdateSource {
private static final byte TYPE_RAW_STRING = -1;
@@ -1010,6 +1011,21 @@ public class InsertRowNode extends InsertNode implements
WALEntryValue {
return new TimeValuePair(time,
TsPrimitiveType.getByType(dataTypes[columnIndex], value));
}
+ @Override
+ public long getLastCacheTimestamp() {
+ return time;
+ }
+
+ @Override
+ public boolean hasLastCacheValue(final int index) {
+ return canComposeTimeValuePair(index);
+ }
+
+ @Override
+ public TimeValuePair getLastCacheValue(final int index) {
+ return composeTimeValuePair(index);
+ }
+
private boolean canComposeTimeValuePair(final int columnIndex) {
return measurements != null
&& columnIndex >= 0
@@ -1026,18 +1042,9 @@ public class InsertRowNode extends InsertNode implements
WALEntryValue {
}
public void updateLastCache(String databaseName) {
- TimeValuePair[] timeValuePairs = new TimeValuePair[measurements.length];
- for (int i = 0; i < measurements.length; i++) {
- timeValuePairs[i] = composeTimeValuePair(i);
- }
TreeDeviceSchemaCacheManager.getInstance()
.updateLastCacheIfExists(
- databaseName,
- getDeviceID(),
- measurements,
- timeValuePairs,
- isAligned,
- measurementSchemas);
+ databaseName, getDeviceID(), measurements, this, isAligned,
measurementSchemas);
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowNode.java
index b11c0f6784e..f0d0d8de7d6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowNode.java
@@ -35,7 +35,6 @@ import
org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferVie
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.IDeviceID.Factory;
-import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -281,13 +280,9 @@ public class RelationalInsertRowNode extends InsertRowNode
{
@Override
public void updateLastCache(String databaseName) {
- TimeValuePair[] timeValuePairs = new TimeValuePair[measurements.length];
- for (int i = 0; i < measurements.length; i++) {
- timeValuePairs[i] = composeTimeValuePair(i);
- }
TableDeviceSchemaCache.getInstance()
.updateLastCacheIfExists(
- databaseName, getDeviceID(), measurements, measurementSchemas,
timeValuePairs);
+ databaseName, getDeviceID(), measurements, measurementSchemas,
this);
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/LastCacheUpdateSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/LastCacheUpdateSource.java
new file mode 100644
index 00000000000..2cd45460024
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/LastCacheUpdateSource.java
@@ -0,0 +1,37 @@
+/*
+ * 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.queryengine.plan.relational.metadata.fetcher.cache;
+
+import org.apache.tsfile.read.TimeValuePair;
+
+/**
+ * Provides row values lazily when updating last cache on the write path.
+ *
+ * <p>{@link #getLastCacheValue(int)} is called only when the corresponding
cache entry exists and
+ * its timestamp is eligible for update.
+ */
+public interface LastCacheUpdateSource {
+
+ long getLastCacheTimestamp();
+
+ boolean hasLastCacheValue(int index);
+
+ TimeValuePair getLastCacheValue(int index);
+}
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 5ea51a5809f..394cd509571 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
@@ -140,13 +140,22 @@ public class TableDeviceCacheEntry {
return 0;
}
// Safe here because tree schema is invalidated by the whole entry
- final int result =
- (deviceSchema.compareAndSet(null, new TreeDeviceNormalSchema(database,
isAligned))
- ? TreeDeviceNormalSchema.INSTANCE_SIZE
- : 0);
- return deviceSchema.get() instanceof TreeDeviceNormalSchema
- ? result + ((TreeDeviceNormalSchema)
deviceSchema.get()).update(measurements, schemas)
- : 0;
+ 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() {
@@ -203,6 +212,18 @@ public class TableDeviceCacheEntry {
return Objects.nonNull(lastCache.get()) ? result : 0;
}
+ int tryUpdateLastCache(
+ final String[] measurements,
+ final @Nullable IMeasurementSchema[] measurementSchemas,
+ final LastCacheUpdateSource updateSource) {
+ final TableDeviceLastCache cache = lastCache.get();
+ final int result =
+ Objects.nonNull(cache)
+ ? cache.tryUpdate(measurements, measurementSchemas, updateSource)
+ : 0;
+ return Objects.nonNull(lastCache.get()) ? result : 0;
+ }
+
int tryUpdateLastCache(final String[] measurements, final TimeValuePair[]
timeValuePairs) {
return tryUpdateLastCache(measurements, timeValuePairs, false);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
index 325484f6f68..5c79457bada 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
@@ -33,11 +33,11 @@ import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import java.util.List;
-import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -73,8 +73,9 @@ public class TableDeviceLastCache {
new TimeValuePair(Long.MIN_VALUE, PLACEHOLDER_NO_VALUE);
// Time is seen as "" as a measurement
- private final Map<String, TimeValuePair> measurement2CachedLastMap = new
ConcurrentHashMap<>();
- private final Map<String, Long> measurement2CachedLastKnownNullTimeMap =
+ private final ConcurrentMap<String, TimeValuePair> measurement2CachedLastMap
=
+ new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, Long>
measurement2CachedLastKnownNullTimeMap =
new ConcurrentHashMap<>();
private final boolean isTableModel;
@@ -143,7 +144,7 @@ public class TableDeviceLastCache {
final @Nullable IMeasurementSchema[] measurementSchemas,
final @Nonnull TimeValuePair[] timeValuePairs,
final boolean invalidateNull) {
- final AtomicInteger diff = new AtomicInteger(0);
+ int diff = 0;
long lastTime = Long.MIN_VALUE;
for (int i = 0; i < measurements.length; ++i) {
@@ -151,49 +152,93 @@ public class TableDeviceLastCache {
if (Objects.isNull(measurement)) {
continue;
}
- if (Objects.isNull(timeValuePairs[i])) {
+ final TimeValuePair timeValuePair = timeValuePairs[i];
+ if (Objects.isNull(timeValuePair)) {
if (invalidateNull) {
- diff.addAndGet(removeKnownNullTime(measurement));
- diff.addAndGet(
- -((int) RamUsageEstimator.sizeOf(measurement)
- +
getTvPairEntrySize(measurement2CachedLastMap.remove(measurement))));
+ diff += removeKnownNullTime(measurement);
+ diff -=
+ (int) RamUsageEstimator.sizeOf(measurement)
+ +
getTvPairEntrySize(measurement2CachedLastMap.remove(measurement));
}
continue;
}
- if (isKnownNullAtAlignedTime(measurement, timeValuePairs[i])) {
- if (lastTime < timeValuePairs[i].getTimestamp()) {
- lastTime = timeValuePairs[i].getTimestamp();
- }
- diff.addAndGet(tryUpdateKnownNullTime(measurement,
timeValuePairs[i].getTimestamp()));
+ if (lastTime < timeValuePair.getTimestamp()) {
+ lastTime = timeValuePair.getTimestamp();
+ }
+ if (isKnownNullAtAlignedTime(measurement, timeValuePair)) {
+ diff += tryUpdateKnownNullTime(measurement,
timeValuePair.getTimestamp());
+ } else {
+ diff += tryUpdateCachedLast(measurement, timeValuePair);
+ }
+ }
+ tryUpdateLastTime(lastTime);
+ return diff;
+ }
+
+ int tryUpdate(
+ final @Nonnull String[] measurements,
+ final @Nullable IMeasurementSchema[] measurementSchemas,
+ final LastCacheUpdateSource updateSource) {
+ int diff = 0;
+ boolean hasValue = false;
+
+ for (int i = 0; i < measurements.length; ++i) {
+ final String measurement = getRawMeasurement(measurements,
measurementSchemas, i);
+ if (Objects.isNull(measurement) || !updateSource.hasLastCacheValue(i)) {
continue;
}
+ hasValue = true;
+ diff += tryUpdateCachedLast(measurement, updateSource, i);
+ }
+ if (hasValue) {
+ tryUpdateLastTime(updateSource.getLastCacheTimestamp());
+ }
+ return diff;
+ }
- final int finalI = i;
- if (lastTime < timeValuePairs[i].getTimestamp()) {
- lastTime = timeValuePairs[i].getTimestamp();
+ private int tryUpdateCachedLast(
+ final String measurement, final LastCacheUpdateSource updateSource,
final int index) {
+ final TimeValuePair cachedPair =
measurement2CachedLastMap.get(measurement);
+ if (Objects.isNull(cachedPair)
+ || cachedPair.getTimestamp() > updateSource.getLastCacheTimestamp()) {
+ return 0;
+ }
+ final TimeValuePair newPair = updateSource.getLastCacheValue(index);
+ return Objects.nonNull(newPair) ? tryUpdateCachedLast(measurement,
cachedPair, newPair) : 0;
+ }
+
+ private int tryUpdateCachedLast(final String measurement, final
TimeValuePair newPair) {
+ return tryUpdateCachedLast(measurement,
measurement2CachedLastMap.get(measurement), newPair);
+ }
+
+ private int tryUpdateCachedLast(
+ final String measurement, TimeValuePair cachedPair, final TimeValuePair
newPair) {
+ while (Objects.nonNull(cachedPair) && cachedPair.getTimestamp() <=
newPair.getTimestamp()) {
+ if (measurement2CachedLastMap.replace(measurement, cachedPair, newPair))
{
+ return getDiffSize(cachedPair, newPair)
+ + clearKnownNullTimeIfCovered(measurement, newPair.getTimestamp());
}
- measurement2CachedLastMap.computeIfPresent(
- measurement,
- (measurementName, tvPair) -> {
- if (tvPair.getTimestamp() <=
timeValuePairs[finalI].getTimestamp()) {
- diff.addAndGet(
- getDiffSize(tvPair, timeValuePairs[finalI])
- + clearKnownNullTimeIfCovered(
- measurementName,
timeValuePairs[finalI].getTimestamp()));
- return timeValuePairs[finalI];
- }
- return tvPair;
- });
+ cachedPair = measurement2CachedLastMap.get(measurement);
+ }
+ return 0;
+ }
+
+ private void tryUpdateLastTime(final long lastTime) {
+ if (lastTime == Long.MIN_VALUE) {
+ return;
+ }
+ TimeValuePair cachedPair = measurement2CachedLastMap.get("");
+ if (Objects.isNull(cachedPair) || cachedPair.getTimestamp() >= lastTime) {
+ return;
+ }
+ final TimeValuePair newPair = new TimeValuePair(lastTime,
PLACEHOLDER_NO_VALUE);
+ while (Objects.nonNull(cachedPair) && cachedPair.getTimestamp() <
lastTime) {
+ if (measurement2CachedLastMap.replace("", cachedPair, newPair)) {
+ return;
+ }
+ cachedPair = measurement2CachedLastMap.get("");
}
- final long finalLastTime = lastTime;
- measurement2CachedLastMap.computeIfPresent(
- "",
- (time, tvPair) ->
- tvPair.getTimestamp() < finalLastTime
- ? new TimeValuePair(finalLastTime, PLACEHOLDER_NO_VALUE)
- : tvPair);
- return diff.get();
}
@Nullable
@@ -360,8 +405,9 @@ public class TableDeviceLastCache {
return 0;
}
final Long knownNullTime =
measurement2CachedLastKnownNullTimeMap.get(measurement);
- if (knownNullTime != null && knownNullTime <= coveredTime) {
- measurement2CachedLastKnownNullTimeMap.remove(measurement);
+ if (knownNullTime != null
+ && knownNullTime <= coveredTime
+ && measurement2CachedLastKnownNullTimeMap.remove(measurement,
knownNullTime)) {
return -getKnownNullTimeEntrySize();
}
return 0;
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 f712b668689..15bd7ab6eca 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
@@ -305,6 +305,20 @@ public class TableDeviceSchemaCache {
false);
}
+ public void updateLastCacheIfExists(
+ final String database,
+ final IDeviceID deviceId,
+ final String[] measurements,
+ final @Nullable IMeasurementSchema[] measurementSchemas,
+ final LastCacheUpdateSource updateSource) {
+ dualKeyCache.update(
+ new TableId(database, deviceId.getTableName()),
+ deviceId,
+ null,
+ entry -> entry.tryUpdateLastCache(measurements, measurementSchemas,
updateSource),
+ false);
+ }
+
/**
* Update the last cache in writing or the second push of last cache query.
If a measurement is
* with all {@code null}s or is a tag/attribute column, its {@link
TimeValuePair}[] shall be
@@ -457,7 +471,7 @@ public class TableDeviceSchemaCache {
dualKeyCache.update(
new TableId(null, deviceID.getTableName()),
deviceID,
- new TableDeviceCacheEntry(),
+ Objects.isNull(timeValuePairs) ? new TableDeviceCacheEntry() : null,
initOrInvalidate
? entry ->
entry.setMeasurementSchema(
@@ -475,6 +489,26 @@ public class TableDeviceSchemaCache {
Objects.isNull(timeValuePairs));
}
+ void updateLastCache(
+ final String database,
+ 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),
+ false);
+ }
+
public boolean getLastCache(
final Map<TableId, Map<IDeviceID, Map<String, Pair<TSDataType,
TimeValuePair>>>> inputMap) {
return dualKeyCache.batchApply(inputMap,
TableDeviceCacheEntry::updateInputMap);
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 8cabd8ca43f..799f938edd4 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
@@ -353,6 +353,17 @@ public class TreeDeviceSchemaCacheManager {
database, deviceID, measurements, timeValuePairs, isAligned,
measurementSchemas, false);
}
+ public void updateLastCacheIfExists(
+ final String database,
+ final IDeviceID deviceID,
+ final String[] measurements,
+ final LastCacheUpdateSource updateSource,
+ final boolean isAligned,
+ final IMeasurementSchema[] measurementSchemas) {
+ tableDeviceSchemaCache.updateLastCache(
+ database, deviceID, measurements, updateSource, isAligned,
measurementSchemas);
+ }
+
/**
* Update the {@link TableDeviceLastCache} on query in tree model.
*
