You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/09/15 10:39:13 UTC
(iotdb) branch try_remove_flush_pipeline created (now d2d53246cc5)
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a change to branch try_remove_flush_pipeline
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at d2d53246cc5 tmp
This branch includes the following new commits:
new d2d53246cc5 tmp
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: tmp
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch try_remove_flush_pipeline
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d2d53246cc5177f55ce9e0beaf12325bd0d9b6d8
Author: Tian Jiang <jt...@163.com>
AuthorDate: Tue Sep 15 18:38:54 2026 +0800
tmp
---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 10 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 5 +
.../dataregion/flush/MemTableFlushTask.java | 64 +++++-
.../FlushSubTaskPipelinePerformanceTest.java | 230 +++++++++++++++++++++
.../MemTableFlushTaskSingleThreadTest.java | 49 +++++
.../conf/iotdb-system.properties.template | 5 +
6 files changed, 360 insertions(+), 3 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 80f9f1f1b30..f826fa1dd02 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -869,6 +869,8 @@ public class IoTDBConfig {
/** the size of ioTaskQueue */
private int ioTaskQueueSizeForFlushing = 10;
+ private boolean enableFlushSubTaskPipeline = true;
+
/** the number of data regions per user-defined database */
private int dataRegionNum = 1;
@@ -3026,6 +3028,14 @@ public class IoTDBConfig {
this.ioTaskQueueSizeForFlushing = ioTaskQueueSizeForFlushing;
}
+ public boolean isEnableFlushSubTaskPipeline() {
+ return enableFlushSubTaskPipeline;
+ }
+
+ public void setEnableFlushSubTaskPipeline(boolean enableFlushSubTaskPipeline) {
+ this.enableFlushSubTaskPipeline = enableFlushSubTaskPipeline;
+ }
+
public boolean isEnableSeqSpaceCompaction() {
return enableSeqSpaceCompaction;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 565f0b7832a..e6090c3d141 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -417,6 +417,11 @@ public class IoTDBDescriptor {
properties.getProperty(
"io_task_queue_size_for_flushing",
Integer.toString(conf.getIoTaskQueueSizeForFlushing()))));
+ conf.setEnableFlushSubTaskPipeline(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "enable_flush_sub_task_pipeline",
+ Boolean.toString(conf.isEnableFlushSubTaskPipeline()))));
boolean enableWALCompression =
Boolean.parseBoolean(properties.getProperty("enable_wal_compression", "true"));
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java
index 68727f03aa7..1ed70bf608e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java
@@ -78,7 +78,7 @@ public class MemTableFlushTask {
private final BlockingQueue<Object> encodingTaskQueue = new LinkedBlockingQueue<>();
private final BlockingQueue<Object> ioTaskQueue =
- (SystemInfo.getInstance().isEncodingFasterThanIo())
+ (config.isEnableFlushSubTaskPipeline() && SystemInfo.getInstance().isEncodingFasterThanIo())
? new LinkedBlockingQueue<>(config.getIoTaskQueueSizeForFlushing())
: new LinkedBlockingQueue<>();
@@ -107,8 +107,10 @@ public class MemTableFlushTask {
this.writer = writer;
this.storageGroup = storageGroup;
this.dataRegionId = dataRegionId;
- this.encodingTaskFuture = SUB_TASK_POOL_MANAGER.submit(encodingTask);
- this.ioTaskFuture = SUB_TASK_POOL_MANAGER.submit(ioTask);
+ this.encodingTaskFuture =
+ config.isEnableFlushSubTaskPipeline() ? SUB_TASK_POOL_MANAGER.submit(encodingTask) : null;
+ this.ioTaskFuture =
+ config.isEnableFlushSubTaskPipeline() ? SUB_TASK_POOL_MANAGER.submit(ioTask) : null;
long MAX_NUMBER_OF_POINTS_IN_CHUNK = config.getTargetChunkPointNum();
long TARGET_CHUNK_SIZE = config.getTargetChunkSize();
@@ -130,6 +132,10 @@ public class MemTableFlushTask {
/** the function for flushing memtable. */
@SuppressWarnings("squid:S3776")
public void syncFlushMemTable() throws ExecutionException, InterruptedException {
+ if (!config.isEnableFlushSubTaskPipeline()) {
+ syncFlushMemTableInSingleThread();
+ return;
+ }
long avgSeriesPointsNum =
memTable.getSeriesNumber() == 0
? 0
@@ -233,6 +239,58 @@ public class MemTableFlushTask {
"flush");
}
+ private void syncFlushMemTableInSingleThread() throws ExecutionException, InterruptedException {
+ Map<IDeviceID, IWritableMemChunkGroup> memTableMap = memTable.getMemTableMap();
+ List<IDeviceID> deviceIDList = new ArrayList<>(memTableMap.keySet());
+ Collections.sort(deviceIDList);
+ for (IDeviceID deviceID : deviceIDList) {
+ IWritableMemChunkGroup group = memTableMap.get(deviceID);
+ if (group.isEmpty() || group.getMemChunkMap().isEmpty()) {
+ continue;
+ }
+ try {
+ writer.startChunkGroup(deviceID);
+ } catch (IOException e) {
+ throw new ExecutionException(e);
+ }
+ List<String> seriesInOrder = new ArrayList<>(group.getMemChunkMap().keySet());
+ Collections.sort(seriesInOrder);
+ for (String seriesId : seriesInOrder) {
+ IWritableMemChunk series = group.getMemChunkMap().get(seriesId);
+ if (series.count() == 0) {
+ continue;
+ }
+ series.sortTvListForFlush();
+ if (series instanceof AlignedWritableMemChunk && times == null) {
+ times = new long[MAX_NUMBER_OF_POINTS_IN_PAGE];
+ }
+ series.encode(ioTaskQueue, encodeInfo, times);
+ Object message;
+ while ((message = ioTaskQueue.poll()) != null) {
+ if (message instanceof IChunkWriter) {
+ try {
+ ((IChunkWriter) message).writeToFileWriter(writer);
+ } catch (IOException e) {
+ throw new ExecutionException(e);
+ }
+ }
+ }
+ }
+ try {
+ writer.setMinPlanIndex(memTable.getMinPlanIndex());
+ writer.setMaxPlanIndex(memTable.getMaxPlanIndex());
+ writer.endChunkGroup();
+ } catch (IOException e) {
+ throw new ExecutionException(e);
+ }
+ }
+ try {
+ writer.writePlanIndices();
+ } catch (IOException e) {
+ throw new ExecutionException(e);
+ }
+ }
+
/** encoding task (second task of pipeline) */
private Runnable encodingTask =
new Runnable() {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlushSubTaskPipelinePerformanceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlushSubTaskPipelinePerformanceTest.java
new file mode 100644
index 00000000000..c781e630d3b
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlushSubTaskPipelinePerformanceTest.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.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.storageengine.dataregion.flush.MemTableFlushTask;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.db.utils.constant.TestConstant;
+
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.apache.tsfile.write.writer.RestorableTsFileIOWriter;
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertTrue;
+
+/** Manual benchmark for comparing flush pipeline and single-thread modes under concurrency. */
+public class FlushSubTaskPipelinePerformanceTest {
+
+ private static final String ENABLED_PROPERTY = "iotdb.flush-sub-task.perf.enabled";
+ private static final String ROUNDS_PROPERTY = "iotdb.flush-sub-task.perf.rounds";
+ private static final String ROWS_PROPERTY = "iotdb.flush-sub-task.perf.rows";
+ private static final String COLUMNS_PROPERTY = "iotdb.flush-sub-task.perf.columns";
+
+ /** Compatibility alias for the previous single-row-count benchmark. */
+ private static final String WORK_PROPERTY = "iotdb.flush-sub-task.perf.work";
+
+ /**
+ * Flush real multi-series INT32 MemTables across row/column sizes and 1-64 concurrent callers in
+ * both modes. Defaults are 10,000 and 100,000 rows, 1 and 4 columns, and three measured rounds
+ * after one warmup round.
+ */
+ @Test
+ public void benchmarkFlushModesByConcurrency() throws Exception {
+ Assume.assumeTrue(Boolean.getBoolean(ENABLED_PROPERTY));
+ boolean originalPipeline =
+ IoTDBDescriptor.getInstance().getConfig().isEnableFlushSubTaskPipeline();
+ EnvironmentUtils.envSetUp();
+ try {
+ int rounds = Integer.getInteger(ROUNDS_PROPERTY, 3);
+ int[] rows = getMatrix(ROWS_PROPERTY, WORK_PROPERTY, new int[] {10_000, 100_000});
+ int[] columns = getMatrix(COLUMNS_PROPERTY, null, new int[] {1, 4});
+ assertTrue(rounds > 0);
+ System.out.printf(
+ Locale.ROOT,
+ "Flush settings rows=%s columns=%s rounds=%d processors=%d subTaskPool=cached%n",
+ formatMatrix(rows),
+ formatMatrix(columns),
+ rounds,
+ Runtime.getRuntime().availableProcessors());
+ for (int rowCount : rows) {
+ for (int columnCount : columns) {
+ for (int concurrency : new int[] {1, 2, 4, 8, 16, 32, 64}) {
+ runMatrixCase(rounds, rowCount, columnCount, concurrency);
+ }
+ }
+ }
+ } finally {
+ IoTDBDescriptor.getInstance().getConfig().setEnableFlushSubTaskPipeline(originalPipeline);
+ EnvironmentUtils.cleanEnv();
+ }
+ }
+
+ private static void runMatrixCase(int rounds, int rows, int columns, int concurrency)
+ throws Exception {
+ long[] elapsedNanos = new long[2];
+ long[] flushNanos = new long[2];
+ // Warm both modes, then alternate their order to reduce systematic warmup bias.
+ for (int round = -1; round < rounds; round++) {
+ for (boolean pipeline :
+ round % 2 == 0 ? new boolean[] {false, true} : new boolean[] {true, false}) {
+ IoTDBDescriptor.getInstance().getConfig().setEnableFlushSubTaskPipeline(pipeline);
+ int mode = pipeline ? 0 : 1;
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<Future<Long>> futures = new ArrayList<>(concurrency);
+ long start = System.nanoTime();
+ try {
+ for (int task = 0; task < concurrency; task++) {
+ int currentTask = task;
+ int currentRound = round;
+ futures.add(
+ executor.submit(
+ () ->
+ runFlushWorkload(
+ pipeline, concurrency, currentRound, rows, columns, currentTask)));
+ }
+ long totalFlushNanos = 0;
+ for (Future<Long> future : futures) {
+ totalFlushNanos += future.get();
+ }
+ long batchNanos = System.nanoTime() - start;
+ if (round >= 0) {
+ elapsedNanos[mode] += batchNanos;
+ flushNanos[mode] += totalFlushNanos;
+ System.out.printf(
+ Locale.ROOT,
+ "Flush sample rows=%d columns=%d mode=%s concurrency=%d round=%d batch=%.3f ms flushMean=%.3f ms%n",
+ rows,
+ columns,
+ pipeline ? "pipeline" : "single-thread",
+ concurrency,
+ round + 1,
+ batchNanos / 1_000_000.0,
+ totalFlushNanos / (double) concurrency / 1_000_000.0);
+ }
+ } finally {
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(1, TimeUnit.MINUTES));
+ }
+ }
+ }
+ for (int mode = 0; mode < 2; mode++) {
+ System.out.printf(
+ Locale.ROOT,
+ "Flush mode rows=%d columns=%d mode=%s concurrency=%d average=%.3f ms flushMean=%.3f ms pointsPerSecond=%.0f%n",
+ rows,
+ columns,
+ mode == 0 ? "pipeline" : "single-thread",
+ concurrency,
+ elapsedNanos[mode] / (double) rounds / 1_000_000.0,
+ flushNanos[mode] / (double) (rounds * concurrency) / 1_000_000.0,
+ (double) rows * columns * concurrency * rounds * 1_000_000_000.0 / elapsedNanos[mode]);
+ }
+ }
+
+ private static long runFlushWorkload(
+ boolean pipeline, int concurrency, int round, int rows, int columns, int taskId) {
+ PrimitiveMemTable memTable = new PrimitiveMemTable("root.flush_perf", Integer.toString(taskId));
+ String filePath =
+ TestConstant.OUTPUT_DATA_DIR
+ + "flush-performance-"
+ + pipeline
+ + "-r"
+ + rows
+ + "-c"
+ + columns
+ + "-x"
+ + concurrency
+ + "-n"
+ + round
+ + "-t"
+ + taskId
+ + ".tsfile";
+ try {
+ produceData(memTable, rows, columns);
+ try (RestorableTsFileIOWriter writer =
+ new RestorableTsFileIOWriter(FSFactoryProducer.getFSFactory().getFile(filePath))) {
+ long start = System.nanoTime();
+ new MemTableFlushTask(memTable, writer, "root.flush_perf", Integer.toString(taskId))
+ .syncFlushMemTable();
+ return System.nanoTime() - start;
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ } finally {
+ memTable.release();
+ }
+ }
+
+ private static void produceData(PrimitiveMemTable memTable, int rows, int columns)
+ throws IllegalPathException {
+ List<IMeasurementSchema> schemas = new ArrayList<>(columns);
+ for (int column = 0; column < columns; column++) {
+ schemas.add(
+ new MeasurementSchema("s" + column, MemTableTestUtils.dataType0, TSEncoding.PLAIN));
+ }
+ IDeviceID deviceId = DeviceIDFactory.getInstance().getDeviceID(new PartialPath("d0"));
+ for (long row = 1; row <= rows; row++) {
+ Object[] values = new Object[columns];
+ for (int column = 0; column < columns; column++) {
+ values[column] = (int) (row + column);
+ }
+ memTable.write(deviceId, schemas, row, values);
+ }
+ }
+
+ private static int[] getMatrix(String property, String fallbackProperty, int[] defaults) {
+ String value = System.getProperty(property);
+ if (value == null && fallbackProperty != null) {
+ value = System.getProperty(fallbackProperty);
+ }
+ if (value == null) {
+ return defaults;
+ }
+ String[] parts = value.split(",", -1);
+ int[] result = new int[parts.length];
+ for (int i = 0; i < parts.length; i++) {
+ result[i] = Integer.parseInt(parts[i].trim());
+ assertTrue(result[i] > 0);
+ }
+ // Repeated dimensions would reuse output paths and invalidate independent measurements.
+ assertTrue(Arrays.stream(result).distinct().count() == result.length);
+ return result;
+ }
+
+ private static String formatMatrix(int[] values) {
+ return Arrays.toString(values);
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskSingleThreadTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskSingleThreadTest.java
new file mode 100644
index 00000000000..9faba60b52f
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskSingleThreadTest.java
@@ -0,0 +1,49 @@
+/*
+ * 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.db.conf.IoTDBDescriptor;
+
+import org.junit.After;
+import org.junit.Before;
+
+/** Runs the existing flush assertions with the pipeline disabled. */
+public class MemTableFlushTaskSingleThreadTest extends MemTableFlushTaskTest {
+ private boolean originalPipeline;
+
+ /** Select the same-thread path before inherited setup creates the writer and MemTable. */
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ originalPipeline = IoTDBDescriptor.getInstance().getConfig().isEnableFlushSubTaskPipeline();
+ IoTDBDescriptor.getInstance().getConfig().setEnableFlushSubTaskPipeline(false);
+ super.setUp();
+ }
+
+ /** Restore shared configuration even when inherited resource cleanup fails. */
+ @After
+ @Override
+ public void tearDown() throws Exception {
+ try {
+ super.tearDown();
+ } finally {
+ IoTDBDescriptor.getInstance().getConfig().setEnableFlushSubTaskPipeline(originalPipeline);
+ }
+ }
+}
diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
index 5079abe5100..2b08cb45162 100644
--- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
+++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
@@ -943,6 +943,11 @@ check_period_when_insert_blocked=50
# Datatype: int
io_task_queue_size_for_flushing=10
+# Whether flush sort, encoding and IO tasks run as a pipeline. If false, they run in one thread.
+# effectiveMode: restart
+# Datatype: bool
+enable_flush_sub_task_pipeline=true
+
# If true, we will estimate each query's possible memory footprint before executing it and deny it if its estimated memory exceeds current free memory
# effectiveMode: hot_reload
# Datatype: bool