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 94694d92f25 Avoid redundant TsFile table schema registration (#18731)
94694d92f25 is described below
commit 94694d92f25531264ca2373a43e9268be19dfd6b
Author: Caideyipi <[email protected]>
AuthorDate: Wed Sep 30 10:03:32 2026 +0800
Avoid redundant TsFile table schema registration (#18731)
---
.../db/schemaengine/table/DataNodeTableCache.java | 1 +
.../iotdb/db/schemaengine/table/ITableCache.java | 2 +
.../db/storageengine/dataregion/DataRegion.java | 6 +-
.../dataregion/memtable/TsFileProcessor.java | 18 +-
.../dataregion/memtable/TsFileProcessorTest.java | 53 +++++
...FileTableSchemaRegistrationPerformanceTest.java | 230 +++++++++++++++++++++
6 files changed, 303 insertions(+), 7 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
index 23daf268206..807387f8c29 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
@@ -435,6 +435,7 @@ public class DataNodeTableCache implements ITableCache {
}
}
+ @Override
public long getInstanceVersion() {
return instanceVersion.get();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java
index 13b05058955..13193fbff13 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java
@@ -29,6 +29,8 @@ import java.util.Map;
public interface ITableCache {
+ long getInstanceVersion();
+
void init(final byte[] tableInitializationBytes);
void preUpdateTable(final String database, final TsTable table, final
@Nullable String oldName);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index e8da377db6b..802d440cc44 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -98,6 +98,7 @@ import
org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.La
import
org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceSchemaCache;
import
org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TreeDeviceSchemaCacheManager;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
+import org.apache.iotdb.db.schemaengine.table.ITableCache;
import org.apache.iotdb.db.service.SettleService;
import org.apache.iotdb.db.service.metrics.CompactionMetrics;
import org.apache.iotdb.db.service.metrics.DataNodeExceptionMetrics;
@@ -1730,14 +1731,15 @@ public class DataRegion implements IDataRegionForQuery {
private void registerToTsFile(InsertNode node, TsFileProcessor
tsFileProcessor) {
final String tableName = node.getTableName();
if (tableName != null) {
+ final ITableCache tableCache = DataNodeTableCache.getInstance();
tsFileProcessor.registerToTsFile(
tableName,
+ tableCache.getInstanceVersion(),
t -> {
final String database = getDatabaseName();
TsTable tsTable =
- DataNodeTableCache.getInstance()
- .getTable(database, t, false,
LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
+ tableCache.getTable(database, t, false,
LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
if (tsTable == null) {
// There is a high probability that the leader node has been
executed and is currently
// located in the follower node.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 6724d196443..78c48a2702b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -113,6 +113,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
@@ -156,6 +157,8 @@ public class TsFileProcessor {
/** Writer for restore tsfile and flushing. */
private RestorableTsFileIOWriter writer;
+ private final Map<String, Long> registeredTableSchemaVersions = new
ConcurrentHashMap<>();
+
/** Tsfile resource for index this tsfile. */
private final TsFileResource tsFileResource;
@@ -2736,11 +2739,16 @@ public class TsFileProcessor {
}
public void registerToTsFile(
- String tableName, Function<String, TableSchema> tableSchemaFunction) {
- getWriter()
- .getSchema()
- .getTableSchemaMap()
- .put(tableName, tableSchemaFunction.apply(tableName));
+ String tableName, long tableCacheVersion, Function<String, TableSchema>
tableSchemaFunction) {
+ registeredTableSchemaVersions.compute(
+ tableName,
+ (name, registeredVersion) -> {
+ if (registeredVersion != null && registeredVersion ==
tableCacheVersion) {
+ return registeredVersion;
+ }
+ getWriter().getSchema().getTableSchemaMap().put(name,
tableSchemaFunction.apply(name));
+ return tableCacheVersion;
+ });
}
public void writeLock() {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
index f1ca1d597c3..b4117ec121a 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
@@ -54,6 +54,7 @@ import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.external.commons.io.FileUtils;
import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.TableSchema;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.read.TimeValuePair;
@@ -145,6 +146,58 @@ public class TsFileProcessorTest {
config.setTargetChunkSize(defaultTargetChunkSize);
}
+ @Test
+ public void testRegisterToTsFileCachesTableSchemaByVersion() throws
IOException {
+ processor =
+ new TsFileProcessor(
+ storageGroup,
+ SystemFileFactory.INSTANCE.getFile(filePath),
+ sgInfo,
+ this::closeTsFileProcessor,
+ (tsFileProcessor, updateMap, systemFlushTime) -> {},
+ true);
+
+ final AtomicInteger supplierCalls = new AtomicInteger();
+ final AtomicInteger secondTableSupplierCalls = new AtomicInteger();
+ final TableSchema firstSchema = new TableSchema("table1");
+ final TableSchema updatedSchema = new TableSchema("table1");
+ final TableSchema secondTableSchema = new TableSchema("table2");
+
+ processor.registerToTsFile(
+ "table1",
+ 1,
+ tableName -> supplierCalls.incrementAndGet() == 1 ? firstSchema :
updatedSchema);
+ processor.registerToTsFile(
+ "table1",
+ 1,
+ tableName -> supplierCalls.incrementAndGet() == 1 ? firstSchema :
updatedSchema);
+
+ Assert.assertEquals(1, supplierCalls.get());
+ Assert.assertSame(
+ firstSchema,
processor.getWriter().getSchema().getTableSchemaMap().get("table1"));
+
+ processor.registerToTsFile(
+ "table2",
+ 1,
+ tableName -> {
+ secondTableSupplierCalls.incrementAndGet();
+ return secondTableSchema;
+ });
+
+ Assert.assertEquals(1, secondTableSupplierCalls.get());
+ Assert.assertSame(
+ secondTableSchema,
processor.getWriter().getSchema().getTableSchemaMap().get("table2"));
+
+ processor.registerToTsFile(
+ "table1",
+ 2,
+ tableName -> supplierCalls.incrementAndGet() == 1 ? firstSchema :
updatedSchema);
+
+ Assert.assertEquals(2, supplierCalls.get());
+ Assert.assertSame(
+ updatedSchema,
processor.getWriter().getSchema().getTableSchemaMap().get("table1"));
+ }
+
@Test
public void testWriteAndFlush()
throws IOException, WriteProcessException, MetadataException,
ExecutionException {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileTableSchemaRegistrationPerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileTableSchemaRegistrationPerformanceTest.java
new file mode 100644
index 00000000000..ad4b2dc84af
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileTableSchemaRegistrationPerformanceTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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.storageengine.dataregion.memtable;
+
+import org.apache.iotdb.commons.file.SystemFileFactory;
+import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo;
+import org.apache.iotdb.db.storageengine.dataregion.DataRegionTest;
+import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+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.iotdb.db.utils.constant.TestConstant;
+
+import org.apache.tsfile.file.metadata.TableSchema;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.Locale;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
+
+public class TsFileTableSchemaRegistrationPerformanceTest {
+
+ private static final String ENABLED_PROPERTY =
"iotdb.tsfile.table.schema.perf.enabled";
+ private static final String WARMUP_ITERATIONS_PROPERTY =
+ "iotdb.tsfile.table.schema.perf.warmup.iterations";
+ private static final String ITERATIONS_PROPERTY =
"iotdb.tsfile.table.schema.perf.iterations";
+ private static final String ROUNDS_PROPERTY =
"iotdb.tsfile.table.schema.perf.rounds";
+ private static volatile int benchmarkBlackhole;
+
+ @Test
+ public void tableSchemaRegistrationBenchmark() throws Exception {
+ Assume.assumeTrue(
+ String.format(
+ "Manual performance UT. Enable with -D%s=true, optionally tune
-D%s, -D%s and -D%s.",
+ ENABLED_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 warmupIterations =
Integer.getInteger(WARMUP_ITERATIONS_PROPERTY, 20_000);
+ final int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 100_000);
+ final int rounds = Integer.getInteger(ROUNDS_PROPERTY, 7);
+ Assert.assertTrue(warmupIterations > 0);
+ Assert.assertTrue(iterations > 0);
+ Assert.assertTrue(rounds > 0);
+
+ final String storageGroup = "root.vehicle";
+ final String systemDir = TestConstant.OUTPUT_DATA_DIR.concat("info");
+ final String filePath = TestConstant.getTestTsFilePath(storageGroup, 0, 0,
0);
+ TsFileProcessor processor = null;
+ EnvironmentUtils.envSetUp();
+ try {
+ final File file = SystemFileFactory.INSTANCE.getFile(filePath);
+ if (!file.getParentFile().exists()) {
+ Assert.assertTrue(file.getParentFile().mkdirs());
+ }
+ final DataRegionInfo regionInfo =
+ new DataRegionInfo(new DataRegionTest.DummyDataRegion(systemDir,
storageGroup));
+ processor =
+ new TsFileProcessor(
+ storageGroup,
+ file,
+ regionInfo,
+ tsFileProcessor -> {},
+ (tsFileProcessor, updateMap, systemFlushTime) -> {},
+ true);
+ final TsFileProcessorInfo processorInfo = new
TsFileProcessorInfo(regionInfo);
+ processor.setTsFileProcessorInfo(processorInfo);
+ regionInfo.initTsFileProcessorInfo(processor);
+ SystemInfo.getInstance().reportStorageGroupStatus(regionInfo, processor);
+
+ final AtomicLong legacySupplierCalls = new AtomicLong();
+ final AtomicLong cachedSupplierCalls = new AtomicLong();
+ final Function<String, TableSchema> legacySupplier =
+ tableName -> {
+ legacySupplierCalls.incrementAndGet();
+ return new TableSchema(tableName);
+ };
+ final Function<String, TableSchema> cachedSupplier =
+ tableName -> {
+ cachedSupplierCalls.incrementAndGet();
+ return new TableSchema(tableName);
+ };
+
+ runLegacy(processor, warmupIterations, legacySupplier);
+ runCached(processor, warmupIterations, cachedSupplier);
+ legacySupplierCalls.set(0);
+ cachedSupplierCalls.set(0);
+
+ final Measurement[] legacyMeasurements = new Measurement[rounds];
+ final Measurement[] cachedMeasurements = new Measurement[rounds];
+ for (int round = 0; round < rounds; ++round) {
+ if ((round & 1) == 0) {
+ legacyMeasurements[round] = measureLegacy(processor, iterations,
legacySupplier);
+ cachedMeasurements[round] = measureCached(processor, iterations,
cachedSupplier);
+ } else {
+ cachedMeasurements[round] = measureCached(processor, iterations,
cachedSupplier);
+ legacyMeasurements[round] = measureLegacy(processor, iterations,
legacySupplier);
+ }
+ }
+
+ final Summary legacySummary =
+ ManualPerformanceTestUtils.summarize(legacyMeasurements, iterations);
+ final Summary cachedSummary =
+ ManualPerformanceTestUtils.summarize(cachedMeasurements, iterations);
+ Assert.assertEquals((long) iterations * rounds,
legacySupplierCalls.get());
+ Assert.assertEquals(0, cachedSupplierCalls.get());
+ printResult(
+ warmupIterations,
+ iterations,
+ rounds,
+ legacySummary,
+ cachedSummary,
+ legacySupplierCalls.get(),
+ cachedSupplierCalls.get());
+ } finally {
+ if (processor != null) {
+ processor.syncClose();
+ }
+ EnvironmentUtils.cleanEnv();
+ EnvironmentUtils.cleanDir(TestConstant.OUTPUT_DATA_DIR);
+ }
+ }
+
+ private static Measurement measureLegacy(
+ TsFileProcessor processor,
+ int iterations,
+ Function<String, TableSchema> tableSchemaFunction) {
+ return ManualPerformanceTestUtils.measure(
+ iterations, () -> legacyRegister(processor, tableSchemaFunction));
+ }
+
+ private static Measurement measureCached(
+ TsFileProcessor processor,
+ int iterations,
+ Function<String, TableSchema> tableSchemaFunction) {
+ return ManualPerformanceTestUtils.measure(
+ iterations, () -> cachedRegister(processor, tableSchemaFunction));
+ }
+
+ private static void runLegacy(
+ TsFileProcessor processor,
+ int iterations,
+ Function<String, TableSchema> tableSchemaFunction) {
+ for (int i = 0; i < iterations; ++i) {
+ legacyRegister(processor, tableSchemaFunction);
+ }
+ }
+
+ private static void runCached(
+ TsFileProcessor processor,
+ int iterations,
+ Function<String, TableSchema> tableSchemaFunction) {
+ for (int i = 0; i < iterations; ++i) {
+ cachedRegister(processor, tableSchemaFunction);
+ }
+ }
+
+ private static void legacyRegister(
+ TsFileProcessor processor, Function<String, TableSchema>
tableSchemaFunction) {
+ final String tableName = "legacy_table";
+ processor
+ .getWriter()
+ .getSchema()
+ .getTableSchemaMap()
+ .put(tableName, tableSchemaFunction.apply(tableName));
+ benchmarkBlackhole =
processor.getWriter().getSchema().getTableSchemaMap().size();
+ }
+
+ private static void cachedRegister(
+ TsFileProcessor processor, Function<String, TableSchema>
tableSchemaFunction) {
+ processor.registerToTsFile("cached_table", 1, tableSchemaFunction);
+ benchmarkBlackhole =
processor.getWriter().getSchema().getTableSchemaMap().size();
+ }
+
+ private static void printResult(
+ int warmupIterations,
+ int iterations,
+ int rounds,
+ Summary legacySummary,
+ Summary cachedSummary,
+ long legacySupplierCalls,
+ long cachedSupplierCalls) {
+ System.out.printf(
+ Locale.ROOT,
+ "TsFile table schema registration benchmark: warmup=%d, iterations=%d,
rounds=%d%n"
+ + "legacy: cpu=%.2f ns/op, allocation=%.2f bytes/op,
supplierCalls=%d%n"
+ + "cached: cpu=%.2f ns/op, allocation=%.2f bytes/op,
supplierCalls=%d%n"
+ + "reduction: cpu=%.2f%%, allocation=%.2f%%%n",
+ warmupIterations,
+ iterations,
+ rounds,
+ legacySummary.getCpuNanosPerOperation(),
+ legacySummary.getAllocatedBytesPerOperation(),
+ legacySupplierCalls,
+ cachedSummary.getCpuNanosPerOperation(),
+ cachedSummary.getAllocatedBytesPerOperation(),
+ cachedSupplierCalls,
+ reduction(legacySummary.getCpuNanosPerOperation(),
cachedSummary.getCpuNanosPerOperation()),
+ reduction(
+ legacySummary.getAllocatedBytesPerOperation(),
+ cachedSummary.getAllocatedBytesPerOperation()));
+ }
+
+ private static double reduction(double before, double after) {
+ return before == 0 ? 0 : (before - after) * 100 / before;
+ }
+}
