You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2026/08/24 08:21:23 UTC
(iotdb) branch rc/2.0.11 updated: [Performance] Reduce temporary allocations on insert path (#18240)
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 <87...@users.noreply.github.com>
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.
*