This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 37cf3cac19c [Performance] Reduce temporary allocations on insert path
(#18240)
37cf3cac19c is described below
commit 37cf3cac19c4f5aca053bd14b41cdaa3618360f1
Author: Caideyipi <[email protected]>
AuthorDate: Mon Jul 20 10:19:29 2026 +0800
[Performance] Reduce temporary allocations on insert path (#18240)
* Reduce temporary allocations on insert path
* Add opt-in write path performance benchmarks
---
.../resource/memory/InsertNodeMemoryEstimator.java | 99 +++++---
.../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 +
.../cache/TreeDeviceSchemaCacheManagerTest.java | 25 +-
.../InsertNodeMemoryEstimatorPerformanceTest.java | 175 ++++++++++++++
.../cache/TableDeviceLastCachePerformanceTest.java | 265 +++++++++++++++++++++
.../fetcher/cache/TableDeviceLastCacheTest.java | 34 +++
.../iotdb/db/utils/ManualPerformanceTestUtils.java | 186 +++++++++++++++
13 files changed, 963 insertions(+), 100 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 b7b660f638c..f4c9442ce01 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
@@ -28,6 +28,7 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
@@ -155,6 +156,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;
+ // Composite 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();
@@ -248,46 +255,62 @@ public class InsertNodeMemoryEstimator {
private static long sizeOfInsertRowsNode(final InsertRowsNode node) {
final Set<Object> deduplicatedObjects =
- newDeduplicatedObjectSetIfNeeded(node.getInsertRowNodeList());
- 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;
+ acquireDeduplicatedObjectSetIfNeeded(node.getInsertRowNodeList());
+ 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 =
- newDeduplicatedObjectSetIfNeeded(node.getInsertRowNodeList());
- 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;
+ acquireDeduplicatedObjectSetIfNeeded(node.getInsertRowNodeList());
+ 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 =
- newDeduplicatedObjectSetIfNeeded(node.getInsertTabletNodeList());
- 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;
+ acquireDeduplicatedObjectSetIfNeeded(node.getInsertTabletNodeList());
+ 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 =
- newDeduplicatedObjectSetIfNeeded(node.getInsertRowNodeList());
- long size = RELATIONAL_INSERT_ROWS_NODE_SIZE;
- size += calculateFullInsertNodeSize(node, deduplicatedObjects);
- size += sizeOfInsertRowNodeList(node.getInsertRowNodeList(),
deduplicatedObjects);
- size += sizeOfIntegerList(node.getInsertRowNodeIndexList());
- // ignore deviceIDs
- return size;
+ acquireDeduplicatedObjectSetIfNeeded(node.getInsertRowNodeList());
+ 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) {
@@ -775,8 +798,24 @@ public class InsertNodeMemoryEstimator {
return Collections.newSetFromMap(new IdentityHashMap<>());
}
- private static Set<Object> newDeduplicatedObjectSetIfNeeded(final List<?>
children) {
- return children != null && children.size() > 1 ?
newDeduplicatedObjectSet() : null;
+ private static Set<Object> acquireDeduplicatedObjectSetIfNeeded(final
List<?> children) {
+ return children != null && children.size() > 1 ?
REUSABLE_DEDUPLICATED_OBJECTS.get() : null;
+ }
+
+ 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();
+ }
+ }
+
+ @TestOnly
+ static void clearReusableDeduplicatedObjectsForTest() {
+ REUSABLE_DEDUPLICATED_OBJECTS.remove();
}
private static boolean shouldCountObject(
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.
*
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 f9188aea9af..c11fc1865a7 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
@@ -50,6 +50,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_MATCH_PATTERN;
@@ -275,37 +276,49 @@ public class TreeDeviceSchemaCacheManagerTest {
}
@Test
- public void testUpdateLastCacheWithAliasDoesNotCopyMeasurements() throws
IllegalPathException {
+ public void testUpdateLastCacheLazilyWithAlias() throws IllegalPathException
{
final String database = "root.db";
final PartialPath device = new PartialPath("root.db.d_alias");
final MeasurementSchema s1 = new MeasurementSchema("s1", TSDataType.INT32);
+ final MeasurementSchema s2 = new MeasurementSchema("s2", TSDataType.INT32);
final MeasurementPath s1Path = new
MeasurementPath(device.concatNode("s1"), s1);
+ final AtomicInteger composedValueCount = new AtomicInteger();
treeDeviceSchemaCacheManager.declareLastCache(database, s1Path);
final InsertRowNode insertRowNode =
new InsertRowNode(
- new
PlanNodeId("testUpdateLastCacheWithAliasDoesNotCopyMeasurements"),
+ new PlanNodeId("testUpdateLastCacheLazilyWithAlias"),
device,
false,
- new String[] {"alias"},
- new TSDataType[] {TSDataType.INT32},
- new MeasurementSchema[] {s1},
+ new String[] {"alias", "uncachedAlias"},
+ new TSDataType[] {TSDataType.INT32, TSDataType.INT32},
+ new MeasurementSchema[] {s1, s2},
1L,
- new Object[] {1},
+ new Object[] {1, 2},
false) {
@Override
public String[] getRawMeasurements() {
throw new AssertionError("Last cache update should not copy raw
measurements");
}
+
+ @Override
+ public TimeValuePair composeTimeValuePair(final int columnIndex) {
+ composedValueCount.incrementAndGet();
+ return super.composeTimeValuePair(columnIndex);
+ }
};
insertRowNode.updateLastCache(database);
+ Assert.assertEquals(1, composedValueCount.get());
Assert.assertEquals(
new TimeValuePair(1L, new TsPrimitiveType.TsInt(1)),
treeDeviceSchemaCacheManager.getLastCache(
new MeasurementPath(device.getIDeviceID(), "s1")));
+ Assert.assertNull(
+ treeDeviceSchemaCacheManager.getLastCache(
+ new MeasurementPath(device.getIDeviceID(), "s2")));
Assert.assertNull(
treeDeviceSchemaCacheManager.getLastCache(
new MeasurementPath(device.getIDeviceID(), "alias")));
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorPerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorPerformanceTest.java
index d7c01a5b676..655347b37cd 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorPerformanceTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorPerformanceTest.java
@@ -22,7 +22,12 @@ package org.apache.iotdb.db.pipe.resource.memory;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils;
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement;
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -30,6 +35,8 @@ import org.junit.Assert;
import org.junit.Assume;
import org.junit.Test;
+import java.util.Locale;
+
public class InsertNodeMemoryEstimatorPerformanceTest {
private static final String ENABLED_PROPERTY =
@@ -40,6 +47,18 @@ public class InsertNodeMemoryEstimatorPerformanceTest {
"iotdb.pipe.insert.node.memory.estimator.perf.warmup.iterations";
private static final String ITERATIONS_PROPERTY =
"iotdb.pipe.insert.node.memory.estimator.perf.iterations";
+ private static final String REUSE_ENABLED_PROPERTY =
+ "iotdb.pipe.insert.node.memory.estimator.reuse.perf.enabled";
+ private static final String REUSE_ROWS_PROPERTY =
+ "iotdb.pipe.insert.node.memory.estimator.reuse.perf.rows";
+ private static final String REUSE_MEASUREMENTS_PROPERTY =
+ "iotdb.pipe.insert.node.memory.estimator.reuse.perf.measurements";
+ private static final String REUSE_WARMUP_ITERATIONS_PROPERTY =
+ "iotdb.pipe.insert.node.memory.estimator.reuse.perf.warmup.iterations";
+ private static final String REUSE_ITERATIONS_PROPERTY =
+ "iotdb.pipe.insert.node.memory.estimator.reuse.perf.iterations";
+ private static final String REUSE_ROUNDS_PROPERTY =
+ "iotdb.pipe.insert.node.memory.estimator.reuse.perf.rounds";
private static volatile long benchmarkBlackhole;
@@ -71,6 +90,98 @@ public class InsertNodeMemoryEstimatorPerformanceTest {
elapsedNanos / (double) iterations / 1_000_000.0);
}
+ @Test
+ public void compositeInsertRowsReusableSetBenchmark() throws
IllegalPathException {
+ Assume.assumeTrue(
+ String.format(
+ "Manual performance UT. Enable with -D%s=true, optionally tune
-D%s, -D%s, -D%s, -D%s and -D%s.",
+ REUSE_ENABLED_PROPERTY,
+ REUSE_ROWS_PROPERTY,
+ REUSE_MEASUREMENTS_PROPERTY,
+ REUSE_WARMUP_ITERATIONS_PROPERTY,
+ REUSE_ITERATIONS_PROPERTY,
+ REUSE_ROUNDS_PROPERTY),
+ Boolean.getBoolean(REUSE_ENABLED_PROPERTY));
+ Assume.assumeTrue(
+ "Current-thread CPU time and allocation metrics are required.",
+ ManualPerformanceTestUtils.enableThreadMetrics());
+
+ final int rowCount = Integer.getInteger(REUSE_ROWS_PROPERTY, 10);
+ final int measurementCount =
Integer.getInteger(REUSE_MEASUREMENTS_PROPERTY, 500);
+ final int warmupIterations =
Integer.getInteger(REUSE_WARMUP_ITERATIONS_PROPERTY, 1_000);
+ final int iterations = Integer.getInteger(REUSE_ITERATIONS_PROPERTY,
20_000);
+ final int rounds = Integer.getInteger(REUSE_ROUNDS_PROPERTY, 5);
+ Assert.assertTrue(rowCount > 1);
+ Assert.assertTrue(measurementCount > 0);
+ Assert.assertTrue(warmupIterations > 0);
+ Assert.assertTrue(iterations > 0);
+ Assert.assertTrue(rounds > 0);
+
+ final InsertRowsNode insertRowsNode =
createCompositeInsertRowsNode(rowCount, measurementCount);
+ InsertNodeMemoryEstimator.clearReusableDeduplicatedObjectsForTest();
+ final long freshSetEstimate =
InsertNodeMemoryEstimator.sizeOf(insertRowsNode);
+ final long reusedSetEstimate =
InsertNodeMemoryEstimator.sizeOf(insertRowsNode);
+ Assert.assertEquals(freshSetEstimate, reusedSetEstimate);
+
+ for (int i = 0; i < warmupIterations; ++i) {
+ if ((i & 1) == 0) {
+ runFreshSetEstimate(insertRowsNode);
+ runEstimate(insertRowsNode);
+ } else {
+ runEstimate(insertRowsNode);
+ runFreshSetEstimate(insertRowsNode);
+ }
+ }
+
+ final Measurement[] freshSetMeasurements = new Measurement[rounds];
+ final Measurement[] reusedSetMeasurements = new Measurement[rounds];
+ for (int i = 0; i < rounds; ++i) {
+ if ((i & 1) == 0) {
+ freshSetMeasurements[i] =
+ ManualPerformanceTestUtils.measure(
+ iterations,
+
InsertNodeMemoryEstimator::clearReusableDeduplicatedObjectsForTest,
+ () -> runEstimate(insertRowsNode));
+ reusedSetMeasurements[i] =
+ ManualPerformanceTestUtils.measure(iterations, () ->
runEstimate(insertRowsNode));
+ } else {
+ reusedSetMeasurements[i] =
+ ManualPerformanceTestUtils.measure(iterations, () ->
runEstimate(insertRowsNode));
+ freshSetMeasurements[i] =
+ ManualPerformanceTestUtils.measure(
+ iterations,
+
InsertNodeMemoryEstimator::clearReusableDeduplicatedObjectsForTest,
+ () -> runEstimate(insertRowsNode));
+ }
+ }
+ InsertNodeMemoryEstimator.clearReusableDeduplicatedObjectsForTest();
+
+ final Summary freshSetSummary =
+ ManualPerformanceTestUtils.summarize(freshSetMeasurements, iterations);
+ final Summary reusedSetSummary =
+ ManualPerformanceTestUtils.summarize(reusedSetMeasurements,
iterations);
+ System.out.printf(
+ Locale.ROOT,
+ "InsertRows memory-estimator set-reuse benchmark: rows=%d,
measurements=%d, warmups=%d, iterations/round=%d, rounds=%d%n",
+ rowCount,
+ measurementCount,
+ warmupIterations,
+ iterations,
+ rounds);
+ printSummary("legacy", freshSetSummary);
+ printSummary("optimized", reusedSetSummary);
+ System.out.printf(
+ Locale.ROOT,
+ " change: CPU speedup=%.2fx, allocation reduction=%.1f%%, peak-heap
reduction=%.1f%%%n",
+ ratio(
+ freshSetSummary.getCpuNanosPerOperation(),
reusedSetSummary.getCpuNanosPerOperation()),
+ reduction(
+ freshSetSummary.getAllocatedBytesPerOperation(),
+ reusedSetSummary.getAllocatedBytesPerOperation()),
+ reduction(
+ freshSetSummary.getPeakHeapDeltaBytes(),
reusedSetSummary.getPeakHeapDeltaBytes()));
+ }
+
private static long runBenchmark(final InsertTabletNode insertTabletNode,
final int iterations) {
final long startTime = System.nanoTime();
for (int i = 0; i < iterations; ++i) {
@@ -79,6 +190,70 @@ public class InsertNodeMemoryEstimatorPerformanceTest {
return System.nanoTime() - startTime;
}
+ private static void runFreshSetEstimate(final InsertRowsNode insertRowsNode)
{
+ InsertNodeMemoryEstimator.clearReusableDeduplicatedObjectsForTest();
+ runEstimate(insertRowsNode);
+ }
+
+ private static void runEstimate(final InsertRowsNode insertRowsNode) {
+ benchmarkBlackhole = InsertNodeMemoryEstimator.sizeOf(insertRowsNode);
+ }
+
+ private static InsertRowsNode createCompositeInsertRowsNode(
+ final int rowCount, final int measurementCount) throws
IllegalPathException {
+ final String[] measurements = new String[measurementCount];
+ final TSDataType[] dataTypes = new TSDataType[measurementCount];
+ final MeasurementSchema[] measurementSchemas = new
MeasurementSchema[measurementCount];
+ final Object[] values = new Object[measurementCount];
+ for (int i = 0; i < measurementCount; ++i) {
+ measurements[i] = "s" + i;
+ dataTypes[i] = TSDataType.INT32;
+ measurementSchemas[i] = new MeasurementSchema(measurements[i],
TSDataType.INT32);
+ values[i] = i;
+ }
+
+ final PlanNodeId planNodeId = new PlanNodeId("composite-memory-estimator");
+ final PartialPath devicePath = new PartialPath("root.memory_estimator.d1");
+ final InsertRowsNode insertRowsNode = new InsertRowsNode(planNodeId);
+ for (int i = 0; i < rowCount; ++i) {
+ insertRowsNode.addOneInsertRowNode(
+ new InsertRowNode(
+ planNodeId,
+ devicePath,
+ false,
+ measurements,
+ dataTypes,
+ measurementSchemas,
+ i,
+ values,
+ false),
+ i);
+ }
+ insertRowsNode.setTargetPath(devicePath);
+ insertRowsNode.setMeasurements(measurements);
+ insertRowsNode.setDataTypes(dataTypes);
+ insertRowsNode.setMeasurementSchemas(measurementSchemas);
+ return insertRowsNode;
+ }
+
+ private static void printSummary(final String label, final Summary summary) {
+ System.out.printf(
+ Locale.ROOT,
+ " %-10s CPU=%.3f us/op, allocated=%.1f bytes/op, peak heap delta=%.3f
MiB%n",
+ label,
+ summary.getCpuNanosPerOperation() / 1_000.0,
+ summary.getAllocatedBytesPerOperation(),
+ summary.getPeakHeapDeltaBytes() / 1024.0 / 1024.0);
+ }
+
+ private static double ratio(final double baseline, final double optimized) {
+ return optimized == 0 ? Double.POSITIVE_INFINITY : baseline / optimized;
+ }
+
+ private static double reduction(final double baseline, final double
optimized) {
+ return baseline == 0 ? 0 : (baseline - optimized) * 100.0 / baseline;
+ }
+
private static InsertTabletNode createWideInsertTabletNode(final int
columnCount)
throws IllegalPathException {
final String[] measurements = new String[columnCount];
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCachePerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCachePerformanceTest.java
new file mode 100644
index 00000000000..d88169999d3
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCachePerformanceTest.java
@@ -0,0 +1,265 @@
+/*
+ * 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.iotdb.db.utils.ManualPerformanceTestUtils;
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement;
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.TimeValuePair;
+import org.apache.tsfile.utils.TsPrimitiveType;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.util.Locale;
+
+public class TableDeviceLastCachePerformanceTest {
+
+ private static final String ENABLED_PROPERTY =
"iotdb.last.cache.write.perf.enabled";
+ private static final String MEASUREMENTS_PROPERTY =
"iotdb.last.cache.write.perf.measurements";
+ private static final String CACHED_INTERVAL_PROPERTY =
+ "iotdb.last.cache.write.perf.cached.interval";
+ private static final String WARMUP_ITERATIONS_PROPERTY =
+ "iotdb.last.cache.write.perf.warmup.iterations";
+ private static final String ITERATIONS_PROPERTY =
"iotdb.last.cache.write.perf.iterations";
+ private static final String ROUNDS_PROPERTY =
"iotdb.last.cache.write.perf.rounds";
+
+ private static volatile int benchmarkBlackhole;
+
+ @Test
+ public void sparseInitializedMeasurementsBenchmark() {
+ Assume.assumeTrue(
+ String.format(
+ "Manual performance UT. Enable with -D%s=true, optionally tune
-D%s, -D%s, -D%s, -D%s and -D%s.",
+ ENABLED_PROPERTY,
+ MEASUREMENTS_PROPERTY,
+ CACHED_INTERVAL_PROPERTY,
+ WARMUP_ITERATIONS_PROPERTY,
+ ITERATIONS_PROPERTY,
+ ROUNDS_PROPERTY),
+ Boolean.getBoolean(ENABLED_PROPERTY));
+ Assume.assumeTrue(
+ "Current-thread CPU time and allocation metrics are required.",
+ ManualPerformanceTestUtils.enableThreadMetrics());
+
+ final int measurementCount = Integer.getInteger(MEASUREMENTS_PROPERTY,
1_000);
+ final int cachedInterval = Integer.getInteger(CACHED_INTERVAL_PROPERTY,
10);
+ final int warmupIterations =
Integer.getInteger(WARMUP_ITERATIONS_PROPERTY, 1_000);
+ final int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 20_000);
+ final int rounds = Integer.getInteger(ROUNDS_PROPERTY, 5);
+ Assert.assertTrue(measurementCount > 0);
+ Assert.assertTrue(cachedInterval > 0 && cachedInterval <=
measurementCount);
+ Assert.assertTrue(warmupIterations > 0);
+ Assert.assertTrue(iterations > 0);
+ Assert.assertTrue(rounds > 0);
+
+ final Scenario legacy = createScenario(measurementCount, cachedInterval);
+ final Scenario optimized = createScenario(measurementCount,
cachedInterval);
+ runLegacyUpdate(legacy);
+ runOptimizedUpdate(optimized);
+ assertCacheEquals(legacy, optimized);
+
+ for (int i = 0; i < warmupIterations; ++i) {
+ if ((i & 1) == 0) {
+ runLegacyUpdate(legacy);
+ runOptimizedUpdate(optimized);
+ } else {
+ runOptimizedUpdate(optimized);
+ runLegacyUpdate(legacy);
+ }
+ }
+
+ final Measurement[] legacyMeasurements = new Measurement[rounds];
+ final Measurement[] optimizedMeasurements = new Measurement[rounds];
+ for (int i = 0; i < rounds; ++i) {
+ if ((i & 1) == 0) {
+ legacyMeasurements[i] =
+ ManualPerformanceTestUtils.measure(iterations, () ->
runLegacyUpdate(legacy));
+ optimizedMeasurements[i] =
+ ManualPerformanceTestUtils.measure(iterations, () ->
runOptimizedUpdate(optimized));
+ } else {
+ optimizedMeasurements[i] =
+ ManualPerformanceTestUtils.measure(iterations, () ->
runOptimizedUpdate(optimized));
+ legacyMeasurements[i] =
+ ManualPerformanceTestUtils.measure(iterations, () ->
runLegacyUpdate(legacy));
+ }
+ }
+
+ assertCacheEquals(legacy, optimized);
+ final Summary legacySummary =
+ ManualPerformanceTestUtils.summarize(legacyMeasurements, iterations);
+ final Summary optimizedSummary =
+ ManualPerformanceTestUtils.summarize(optimizedMeasurements,
iterations);
+ printResult(
+ measurementCount,
+ legacy.cachedMeasurementCount,
+ warmupIterations,
+ iterations,
+ rounds,
+ legacySummary,
+ optimizedSummary);
+ }
+
+ private static Scenario createScenario(final int measurementCount, final int
cachedInterval) {
+ final String[] measurements = new String[measurementCount];
+ final MeasurementSchema[] measurementSchemas = new
MeasurementSchema[measurementCount];
+ final int[] values = new int[measurementCount];
+ for (int i = 0; i < measurementCount; ++i) {
+ measurements[i] = "s" + i;
+ measurementSchemas[i] = new MeasurementSchema(measurements[i],
TSDataType.INT32);
+ values[i] = i;
+ }
+
+ final int cachedMeasurementCount = (measurementCount + cachedInterval - 1)
/ cachedInterval;
+ final String[] cachedMeasurements = new String[cachedMeasurementCount];
+ for (int i = 0; i < cachedMeasurementCount; ++i) {
+ cachedMeasurements[i] = measurements[i * cachedInterval];
+ }
+
+ final TableDeviceLastCache cache = new TableDeviceLastCache(false);
+ cache.initOrInvalidate(null, null, cachedMeasurements, false);
+ return new Scenario(
+ cache,
+ measurements,
+ measurementSchemas,
+ new RowUpdateSource(values),
+ cachedMeasurementCount);
+ }
+
+ private static void runLegacyUpdate(final Scenario scenario) {
+ // Keep the eager implementation from origin/master for comparison.
+ final TimeValuePair[] timeValuePairs = new
TimeValuePair[scenario.measurements.length];
+ for (int i = 0; i < scenario.measurements.length; ++i) {
+ timeValuePairs[i] = scenario.updateSource.getLastCacheValue(i);
+ }
+ benchmarkBlackhole =
+ scenario.cache.tryUpdate(
+ scenario.measurements, scenario.measurementSchemas,
timeValuePairs, false);
+ }
+
+ private static void runOptimizedUpdate(final Scenario scenario) {
+ benchmarkBlackhole =
+ scenario.cache.tryUpdate(
+ scenario.measurements, scenario.measurementSchemas,
scenario.updateSource);
+ }
+
+ private static void assertCacheEquals(
+ final Scenario expectedScenario, final Scenario actualScenario) {
+ for (final String measurement : expectedScenario.measurements) {
+ Assert.assertEquals(
+ expectedScenario.cache.getTimeValuePair(measurement),
+ actualScenario.cache.getTimeValuePair(measurement));
+ }
+ }
+
+ private static void printResult(
+ final int measurementCount,
+ final int cachedMeasurementCount,
+ final int warmupIterations,
+ final int iterations,
+ final int rounds,
+ final Summary legacy,
+ final Summary optimized) {
+ System.out.printf(
+ Locale.ROOT,
+ "Last-cache row-update benchmark: measurements=%d, initialized=%d,
warmups=%d, iterations/round=%d, rounds=%d%n",
+ measurementCount,
+ cachedMeasurementCount,
+ warmupIterations,
+ iterations,
+ rounds);
+ printSummary("legacy", legacy);
+ printSummary("optimized", optimized);
+ System.out.printf(
+ Locale.ROOT,
+ " change: CPU speedup=%.2fx, allocation reduction=%.1f%%, peak-heap
reduction=%.1f%%%n",
+ ratio(legacy.getCpuNanosPerOperation(),
optimized.getCpuNanosPerOperation()),
+ reduction(
+ legacy.getAllocatedBytesPerOperation(),
optimized.getAllocatedBytesPerOperation()),
+ reduction(legacy.getPeakHeapDeltaBytes(),
optimized.getPeakHeapDeltaBytes()));
+ }
+
+ private static void printSummary(final String label, final Summary summary) {
+ System.out.printf(
+ Locale.ROOT,
+ " %-9s CPU=%.3f us/op, allocated=%.1f bytes/op, peak heap delta=%.3f
MiB%n",
+ label,
+ summary.getCpuNanosPerOperation() / 1_000.0,
+ summary.getAllocatedBytesPerOperation(),
+ summary.getPeakHeapDeltaBytes() / 1024.0 / 1024.0);
+ }
+
+ private static double ratio(final double baseline, final double optimized) {
+ return optimized == 0 ? Double.POSITIVE_INFINITY : baseline / optimized;
+ }
+
+ private static double reduction(final double baseline, final double
optimized) {
+ return baseline == 0 ? 0 : (baseline - optimized) * 100.0 / baseline;
+ }
+
+ private static final class Scenario {
+
+ private final TableDeviceLastCache cache;
+ private final String[] measurements;
+ private final MeasurementSchema[] measurementSchemas;
+ private final RowUpdateSource updateSource;
+ private final int cachedMeasurementCount;
+
+ private Scenario(
+ final TableDeviceLastCache cache,
+ final String[] measurements,
+ final MeasurementSchema[] measurementSchemas,
+ final RowUpdateSource updateSource,
+ final int cachedMeasurementCount) {
+ this.cache = cache;
+ this.measurements = measurements;
+ this.measurementSchemas = measurementSchemas;
+ this.updateSource = updateSource;
+ this.cachedMeasurementCount = cachedMeasurementCount;
+ }
+ }
+
+ private static final class RowUpdateSource implements LastCacheUpdateSource {
+
+ private final int[] values;
+
+ private RowUpdateSource(final int[] values) {
+ this.values = values;
+ }
+
+ @Override
+ public long getLastCacheTimestamp() {
+ return 1L;
+ }
+
+ @Override
+ public boolean hasLastCacheValue(final int index) {
+ return true;
+ }
+
+ @Override
+ public TimeValuePair getLastCacheValue(final int index) {
+ return new TimeValuePair(1L, new TsPrimitiveType.TsInt(values[index]));
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
index eedbbfbb4ec..4ae3228013e 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
@@ -28,9 +28,43 @@ import org.junit.Test;
import java.util.Collections;
import java.util.Optional;
import java.util.OptionalLong;
+import java.util.concurrent.atomic.AtomicInteger;
public class TableDeviceLastCacheTest {
+ @Test
+ public void testLazyUpdateOnlyComposesCachedMeasurements() {
+ final TableDeviceLastCache cache = new TableDeviceLastCache(false);
+ cache.initOrInvalidate(null, null, new String[] {"s1"}, false);
+ final AtomicInteger composedValueCount = new AtomicInteger();
+
+ cache.tryUpdate(
+ new String[] {"s1", "s2"},
+ null,
+ new LastCacheUpdateSource() {
+ @Override
+ public long getLastCacheTimestamp() {
+ return 1L;
+ }
+
+ @Override
+ public boolean hasLastCacheValue(final int index) {
+ return true;
+ }
+
+ @Override
+ public TimeValuePair getLastCacheValue(final int index) {
+ composedValueCount.incrementAndGet();
+ return new TimeValuePair(1L, new TsPrimitiveType.TsInt(index + 1));
+ }
+ });
+
+ Assert.assertEquals(1, composedValueCount.get());
+ Assert.assertEquals(
+ new TimeValuePair(1L, new TsPrimitiveType.TsInt(1)),
cache.getTimeValuePair("s1"));
+ Assert.assertNull(cache.getTimeValuePair("s2"));
+ }
+
@Test
public void testKnownNullTimePreservesHistoricalValueAndClearsOnNewerValue()
{
final TableDeviceLastCache cache = new TableDeviceLastCache(false);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/ManualPerformanceTestUtils.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/ManualPerformanceTestUtils.java
new file mode 100644
index 00000000000..6a489f1afe0
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/ManualPerformanceTestUtils.java
@@ -0,0 +1,186 @@
+/*
+ * 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.utils;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryPoolMXBean;
+import java.lang.management.MemoryType;
+import java.lang.management.MemoryUsage;
+import java.lang.management.ThreadMXBean;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public final class ManualPerformanceTestUtils {
+
+ private static final ThreadMXBean THREAD_MX_BEAN =
ManagementFactory.getThreadMXBean();
+ private static final com.sun.management.ThreadMXBean ALLOCATION_MX_BEAN =
+ THREAD_MX_BEAN instanceof com.sun.management.ThreadMXBean
+ ? (com.sun.management.ThreadMXBean) THREAD_MX_BEAN
+ : null;
+ private static final Runnable NO_OP = () -> {};
+
+ private ManualPerformanceTestUtils() {}
+
+ public static boolean enableThreadMetrics() {
+ if (!THREAD_MX_BEAN.isCurrentThreadCpuTimeSupported()
+ || ALLOCATION_MX_BEAN == null
+ || !ALLOCATION_MX_BEAN.isThreadAllocatedMemorySupported()) {
+ return false;
+ }
+ try {
+ if (!THREAD_MX_BEAN.isThreadCpuTimeEnabled()) {
+ THREAD_MX_BEAN.setThreadCpuTimeEnabled(true);
+ }
+ if (!ALLOCATION_MX_BEAN.isThreadAllocatedMemoryEnabled()) {
+ ALLOCATION_MX_BEAN.setThreadAllocatedMemoryEnabled(true);
+ }
+ final long threadId = Thread.currentThread().getId();
+ return THREAD_MX_BEAN.getCurrentThreadCpuTime() >= 0
+ && ALLOCATION_MX_BEAN.getThreadAllocatedBytes(threadId) >= 0;
+ } catch (final UnsupportedOperationException | SecurityException ignored) {
+ return false;
+ }
+ }
+
+ public static Measurement measure(final int iterations, final Runnable
operation) {
+ return measure(iterations, NO_OP, operation);
+ }
+
+ public static Measurement measure(
+ final int iterations, final Runnable beforeEachIteration, final Runnable
operation) {
+ final List<MemoryPoolMXBean> heapPools = getHeapMemoryPools();
+ System.gc();
+ System.runFinalization();
+ heapPools.forEach(MemoryPoolMXBean::resetPeakUsage);
+ final long baselineHeapBytes = getUsedHeapBytes(heapPools);
+
+ final long threadId = Thread.currentThread().getId();
+ final long allocatedBytesBefore =
ALLOCATION_MX_BEAN.getThreadAllocatedBytes(threadId);
+ long cpuNanos = 0;
+ for (int i = 0; i < iterations; ++i) {
+ beforeEachIteration.run();
+ final long cpuNanosBefore = THREAD_MX_BEAN.getCurrentThreadCpuTime();
+ operation.run();
+ cpuNanos += THREAD_MX_BEAN.getCurrentThreadCpuTime() - cpuNanosBefore;
+ }
+ final long allocatedBytes =
+ ALLOCATION_MX_BEAN.getThreadAllocatedBytes(threadId) -
allocatedBytesBefore;
+ final long peakHeapDeltaBytes = Math.max(0L, getPeakHeapBytes(heapPools) -
baselineHeapBytes);
+ return new Measurement(cpuNanos, allocatedBytes, peakHeapDeltaBytes);
+ }
+
+ public static Summary summarize(final Measurement[] measurements, final int
iterations) {
+ final long[] cpuNanos = new long[measurements.length];
+ final long[] allocatedBytes = new long[measurements.length];
+ final long[] peakHeapDeltaBytes = new long[measurements.length];
+ for (int i = 0; i < measurements.length; ++i) {
+ cpuNanos[i] = measurements[i].cpuNanos;
+ allocatedBytes[i] = measurements[i].allocatedBytes;
+ peakHeapDeltaBytes[i] = measurements[i].peakHeapDeltaBytes;
+ }
+ return new Summary(
+ median(cpuNanos) / iterations,
+ median(allocatedBytes) / iterations,
+ median(peakHeapDeltaBytes));
+ }
+
+ private static List<MemoryPoolMXBean> getHeapMemoryPools() {
+ final List<MemoryPoolMXBean> heapPools = new ArrayList<>();
+ for (final MemoryPoolMXBean memoryPool :
ManagementFactory.getMemoryPoolMXBeans()) {
+ if (memoryPool.getType() == MemoryType.HEAP && memoryPool.isValid()) {
+ heapPools.add(memoryPool);
+ }
+ }
+ return heapPools;
+ }
+
+ private static long getUsedHeapBytes(final List<MemoryPoolMXBean> heapPools)
{
+ long usedHeapBytes = 0;
+ for (final MemoryPoolMXBean heapPool : heapPools) {
+ final MemoryUsage usage = heapPool.getUsage();
+ if (usage != null) {
+ usedHeapBytes += usage.getUsed();
+ }
+ }
+ return usedHeapBytes;
+ }
+
+ private static long getPeakHeapBytes(final List<MemoryPoolMXBean> heapPools)
{
+ long peakHeapBytes = 0;
+ for (final MemoryPoolMXBean heapPool : heapPools) {
+ final MemoryUsage peakUsage = heapPool.getPeakUsage();
+ if (peakUsage != null) {
+ peakHeapBytes += peakUsage.getUsed();
+ }
+ }
+ return peakHeapBytes;
+ }
+
+ private static double median(final long[] values) {
+ Arrays.sort(values);
+ final int middle = values.length / 2;
+ return (values.length & 1) == 1
+ ? values[middle]
+ : values[middle - 1] + (values[middle] - values[middle - 1]) / 2.0;
+ }
+
+ public static final class Measurement {
+
+ private final long cpuNanos;
+ private final long allocatedBytes;
+ private final long peakHeapDeltaBytes;
+
+ private Measurement(
+ final long cpuNanos, final long allocatedBytes, final long
peakHeapDeltaBytes) {
+ this.cpuNanos = cpuNanos;
+ this.allocatedBytes = allocatedBytes;
+ this.peakHeapDeltaBytes = peakHeapDeltaBytes;
+ }
+ }
+
+ public static final class Summary {
+
+ private final double cpuNanosPerOperation;
+ private final double allocatedBytesPerOperation;
+ private final double peakHeapDeltaBytes;
+
+ private Summary(
+ final double cpuNanosPerOperation,
+ final double allocatedBytesPerOperation,
+ final double peakHeapDeltaBytes) {
+ this.cpuNanosPerOperation = cpuNanosPerOperation;
+ this.allocatedBytesPerOperation = allocatedBytesPerOperation;
+ this.peakHeapDeltaBytes = peakHeapDeltaBytes;
+ }
+
+ public double getCpuNanosPerOperation() {
+ return cpuNanosPerOperation;
+ }
+
+ public double getAllocatedBytesPerOperation() {
+ return allocatedBytesPerOperation;
+ }
+
+ public double getPeakHeapDeltaBytes() {
+ return peakHeapDeltaBytes;
+ }
+ }
+}
