From ad71d66d71209f84d68f24c6fff04649d471520a Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 19 Oct 2026 10:00:12 +0800 Subject: [PATCH 1/8] [Test] Use consistent time units for asynchronous assertion flushes --- .../iotdb/db/it/utils/IoTDBTestUtilsIT.java | 7 ++++ .../apache/iotdb/db/it/utils/TestUtils.java | 36 +++++++++++++------ 2 files changed, 32 insertions(+), 11 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java index 80d00c3843b3d..814de53ba68e4 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java @@ -37,6 +37,7 @@ import java.util.function.Consumer; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -66,6 +67,12 @@ public void testAssertAlwaysFail() { () -> TestUtils.assertAlwaysFail(EnvFactory.getEnv(), "show databases", 1)); } + @Test + public void testHalfTimeoutElapsedUsesSeconds() { + assertFalse(TestUtils.isHalfTimeoutElapsed(299_999, 600)); + assertTrue(TestUtils.isHalfTimeoutElapsed(300_001, 600)); + } + @Test public void testAssertDataAlwaysFailsOnMismatch() { final BaseEnv env = EnvFactory.getEnv(); diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java index e559fe9fb1897..79a2f1c984769 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java @@ -85,6 +85,10 @@ public class TestUtils { public static final String TIME_PRECISION_IN_NS = "ns"; + static boolean isHalfTimeoutElapsed(final long elapsedTimeInMillis, final long timeoutSeconds) { + return elapsedTimeInMillis > TimeUnit.SECONDS.toMillis(timeoutSeconds) / 2; + } + /** * Creates a statement on the write node without comparing results across DataNodes. * @@ -1702,7 +1706,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -1783,7 +1789,9 @@ public static void assertDataSizeEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -1822,7 +1830,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -1904,7 +1914,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2024,7 +2036,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2095,8 +2109,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime - > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, consistentSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2137,8 +2151,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime - > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, consistentSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2192,8 +2206,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime - > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, consistentSeconds)) { flushed[0] = true; statement.execute("flush"); } From 3be82a7fee14ff70f57b9313e686bababeea96a3 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 19 Oct 2026 10:01:34 +0800 Subject: [PATCH 2/8] [Test] Trigger Daily IT for TestUtils flush verification --- .github/workflows/daily-it.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/daily-it.yml b/.github/workflows/daily-it.yml index 8d9b74862e4f5..4aa36f5e102ce 100644 --- a/.github/workflows/daily-it.yml +++ b/.github/workflows/daily-it.yml @@ -1,6 +1,9 @@ name: Daily IT on: + pull_request: + branches: + - master schedule: # Run at UTC 19:00 every day (CST 03:00 AM) - cron: "0 19 * * *" From e103b15e02764a5087ee34c02cf28e621ef4e5db Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:11:45 +0800 Subject: [PATCH 3/8] [Fix] Preserve table schemas in consensus LOAD TsFiles Register table schemas before sealing consensus LOAD files, including recovered tasks, so Pipe can enumerate their table data. Cover normal loads, recovery, and null tags with regression tests. --- .../load/TsFileWriterManager.java | 16 ++ .../write/TsFilePrecalculatedChunkWriter.java | 14 ++ .../load/LoadTsFileTableSchemaTest.java | 181 ++++++++++++++++++ 3 files changed, 211 insertions(+) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java index 2c81646c53de3..fc8f38b850288 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java @@ -24,12 +24,16 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.file.SystemFileFactory; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.commons.utils.RetryUtils; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadFileException; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusOp; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -502,6 +506,18 @@ void prepare( fileLength)); } } + // Rebuild from the staged devices so recovered writers also retain the table schemas that + // Pipe needs to enumerate table data in the sealed file. + if (PathUtils.isTableModelDatabase(dataRegion.getDatabaseName())) { + for (final String tableName : writer.getTableNames()) { + final TsTable table = + DataNodeTableCache.getInstance() + .getTable(dataRegion.getDatabaseName(), tableName, false); + if (table != null) { + writer.registerTableSchema(TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table)); + } + } + } writer.close(); final TsFileResource tsFileResource = timePartition2Resource.get(entry.getKey()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java b/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java index a6c0d996c9817..b48b7d14cb17d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java @@ -31,6 +31,7 @@ import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.MeasurementMetadataIndexEntry; import org.apache.tsfile.file.metadata.MetadataIndexNode; +import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.file.metadata.TimeseriesMetadata; import org.apache.tsfile.file.metadata.TsFileMetadata; import org.apache.tsfile.file.metadata.enums.MetadataIndexNodeType; @@ -53,6 +54,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.TreeMap; import java.util.stream.Collectors; @@ -80,6 +82,7 @@ public class TsFilePrecalculatedChunkWriter implements AutoCloseable { private final File file; private final Map>> device2MetadataMap = new TreeMap<>(); + private final Map tableSchemaMap = new TreeMap<>(); private long maxPhysicalFileSize; private boolean sealed; @@ -385,6 +388,7 @@ public void close() throws IOException { } tsFileMetadata.setTableMetadataIndexNodeMap(tableIndexNodeMap); + tsFileMetadata.setTableSchemaMap(tableSchemaMap); tsFileMetadata.setMetaOffset(metaOffset); tsFileMetadata.setBloomFilter(bloomFilter); @@ -446,6 +450,16 @@ public Map> getChunkMetadataListMap() { return result; } + public Set getTableNames() { + return device2MetadataMap.keySet().stream() + .map(IDeviceID::getTableName) + .collect(Collectors.toSet()); + } + + public void registerTableSchema(final TableSchema tableSchema) { + tableSchemaMap.put(tableSchema.getTableName(), tableSchema); + } + private String getFileForLog() { return file == null ? "" : file.getAbsolutePath(); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java new file mode 100644 index 0000000000000..306f3050c4882 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java @@ -0,0 +1,181 @@ +/* + * 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.load; + +import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.parser.table.TsFileInsertionEventTableParser; +import org.apache.iotdb.db.queryengine.plan.scheduler.load.ChunkOffsetCalculator; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.storageengine.dataregion.DataRegion; +import org.apache.iotdb.db.storageengine.load.splitter.ChunkData; +import org.apache.iotdb.db.storageengine.load.splitter.TsFileData; +import org.apache.iotdb.db.storageengine.load.splitter.TsFileSplitter; +import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +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.TsFileSequenceReader; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.record.Tablet; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.Mockito; + +import java.io.File; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import static org.junit.Assert.assertEquals; + +public class LoadTsFileTableSchemaTest { + + private static final String DATABASE = "load_table_schema_test"; + private static final String TABLE = "measurements"; + private static final int ROW_COUNT = 20; + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private String[] originalLoadDirs; + private DataRegion dataRegion; + private TsTable table; + + @Before + public void setUp() throws Exception { + originalLoadDirs = IoTDBDescriptor.getInstance().getConfig().getLoadTsFileDirs(); + IoTDBDescriptor.getInstance() + .getConfig() + .setLoadTsFileDirs(new String[] {temporaryFolder.getRoot().getAbsolutePath()}); + dataRegion = Mockito.mock(DataRegion.class); + Mockito.when(dataRegion.getDatabaseName()).thenReturn(DATABASE); + Mockito.when(dataRegion.getDataRegionIdString()).thenReturn("0"); + Mockito.when(dataRegion.getNonSystemDatabaseName()).thenReturn(Optional.of(DATABASE)); + + table = new TsTable(TABLE); + table.addColumnSchema(new TagColumnSchema("device", TSDataType.STRING)); + table.addColumnSchema( + new FieldColumnSchema("value", TSDataType.INT32, TSEncoding.PLAIN, CompressionType.LZ4)); + DataNodeTableCache.getInstance().preUpdateTable(DATABASE, table, null); + DataNodeTableCache.getInstance().commitUpdateTable(DATABASE, TABLE, null); + } + + @After + public void tearDown() { + DataNodeTableCache.getInstance().invalid(DATABASE); + IoTDBDescriptor.getInstance().getConfig().setLoadTsFileDirs(originalLoadDirs); + } + + @Test + public void testLoadedTableRemainsReadableByPipe() throws Exception { + assertLoadedTableReadable(false, false); + } + + @Test + public void testRecoveredTableRemainsReadableByPipe() throws Exception { + assertLoadedTableReadable(true, false); + } + + @Test + public void testLoadedTableWithNullTagsRemainsReadableByPipe() throws Exception { + assertLoadedTableReadable(false, true); + } + + private void assertLoadedTableReadable(final boolean recover, final boolean nullTags) + throws Exception { + final File source = new File(temporaryFolder.getRoot(), "source.tsfile"); + final TableSchema schema = TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table); + try (final TsFileWriter writer = new TsFileWriter(source)) { + writer.registerTableSchema(schema); + final Tablet tablet = + new Tablet( + TABLE, + Arrays.asList("device", "value"), + Arrays.asList(TSDataType.STRING, TSDataType.INT32), + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD), + ROW_COUNT); + for (int i = 0; i < ROW_COUNT; i++) { + tablet.addTimestamp(i, i); + tablet.addValue("device", i, nullTags ? null : "d" + i % 2); + tablet.addValue("value", i, i); + } + writer.writeTable(tablet); + } + + final File taskDir = temporaryFolder.newFolder("task"); + TsFileWriterManager manager = new TsFileWriterManager(dataRegion, taskDir); + try { + final ChunkOffsetCalculator calculator = new ChunkOffsetCalculator(); + final List pieces = new ArrayList<>(); + new TsFileSplitter( + source, + data -> { + calculator.assign((ChunkData) data); + pieces.add(data); + return true; + }) + .splitTsFileByDataPartition(); + manager.writePiece(pieces); + if (recover) { + manager.close(true); + manager = new TsFileWriterManager(dataRegion, taskDir).recoverFromDisk(); + } + manager.prepare(false, Collections.emptyMap(), Collections.emptyMap()); + + final File staged = new File(taskDir, DATABASE + "-0-0.tsfile"); + int rowCount = 0; + try (final TsFileInsertionEventTableParser parser = + new TsFileInsertionEventTableParser( + staged, + new TablePattern(true, null, null), + Long.MIN_VALUE, + Long.MAX_VALUE, + null, + null, + null, + false)) { + for (final TabletInsertionEvent event : parser.toTabletInsertionEvents()) { + final PipeRawTabletInsertionEvent tabletEvent = (PipeRawTabletInsertionEvent) event; + rowCount += tabletEvent.convertToTablet().getRowSize(); + tabletEvent.clearReferenceCount(getClass().getName()); + } + } + assertEquals(ROW_COUNT, rowCount); + try (final TsFileSequenceReader reader = new TsFileSequenceReader(staged.getPath())) { + assertEquals(schema, reader.getTableSchemaMap().get(TABLE)); + } + } finally { + manager.close(); + } + } +} From bf4492ad19d6ec7c46996caad692263f04d42d92 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 19 Oct 2026 15:16:33 +0800 Subject: [PATCH 4/8] [Fix] Pause ConfigNode Pipe events polled after stop --- .../agent/task/PipeConfigNodeSubtask.java | 12 +- .../agent/task/PipeConfigNodeSubtaskTest.java | 150 ++++++++++++++++++ 2 files changed, 159 insertions(+), 3 deletions(-) create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtaskTest.java diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java index 95376af638b1a..9162c792c5adc 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java @@ -176,16 +176,22 @@ private void initSink(final Map sinkAttributes) throws Exception */ @Override protected boolean executeOnce() throws Exception { - if (isClosed.get()) { + if (isClosed.get() || shouldStopSubmittingSelf.get()) { return false; } final Event event = lastEvent != null ? lastEvent : source.supply(); - // Record the last event for retry when exception occurs + // Keep the event for retry or restart if supply() returns after the task is stopped. setLastEvent(event); + // close() may have released the last event before supply() returned. + if (isClosed.get()) { + clearReferenceCountAndReleaseLastEvent(event); + return false; + } + try { - if (event == null) { + if (event == null || shouldStopSubmittingSelf.get()) { return false; } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtaskTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtaskTest.java new file mode 100644 index 0000000000000..2ca2aca5302c4 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtaskTest.java @@ -0,0 +1,150 @@ +/* + * 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.confignode.manager.pipe.agent.task; + +import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; +import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; +import org.apache.iotdb.commons.pipe.agent.task.subtask.PipeAbstractSinkSubtask; +import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.pipe.api.PipeConnector; +import org.apache.iotdb.pipe.api.PipeExtractor; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +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.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyZeroInteractions; +import static org.mockito.Mockito.when; + +public class PipeConfigNodeSubtaskTest { + + private PipeConfigNodeSubtask subtask; + private PipeExtractor source; + private PipeConnector sink; + private boolean subtaskClosed; + + @Before + public void setUp() throws Exception { + source = mock(PipeExtractor.class); + sink = mock(PipeConnector.class); + subtask = + new PipeConfigNodeSubtask( + "PipeConfigNodeSubtaskTest", + System.currentTimeMillis(), + Collections.emptyMap(), + Collections.emptyMap(), + Collections.singletonMap( + PipeSinkConstant.CONNECTOR_KEY, + BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName()), + new PipeTaskMeta(MinimumProgressIndex.INSTANCE, Integer.MIN_VALUE)); + final Field sourceField = PipeConfigNodeSubtask.class.getDeclaredField("source"); + sourceField.setAccessible(true); + sourceField.set(subtask, source); + final Field sinkField = PipeAbstractSinkSubtask.class.getDeclaredField("outputPipeSink"); + sinkField.setAccessible(true); + sinkField.set(subtask, sink); + } + + @After + public void tearDown() { + if (!subtaskClosed) { + subtask.close(); + } + } + + @Test + public void testStoppedSubtaskDoesNotPollSource() throws Exception { + assertFalse(subtask.executeOnce()); + verifyZeroInteractions(source, sink); + } + + @Test + public void testStopDuringSupplyRetainsEventUntilRestart() throws Exception { + assertStopDuringSupply(false); + } + + @Test + public void testDropDuringSupplyReleasesEvent() throws Exception { + assertStopDuringSupply(true); + } + + private void assertStopDuringSupply(final boolean drop) throws Exception { + final EnrichedEvent event = mock(EnrichedEvent.class); + final CountDownLatch supplyStarted = new CountDownLatch(1); + final CountDownLatch supplyReleased = new CountDownLatch(1); + final ExecutorService worker = Executors.newSingleThreadExecutor(); + when(source.supply()) + .thenAnswer( + invocation -> { + supplyStarted.countDown(); + assertTrue(supplyReleased.await(10, TimeUnit.SECONDS)); + return event; + }); + subtask.allowSubmittingSelf(); + try { + final Future execution = worker.submit(() -> subtask.executeOnce()); + assertTrue(supplyStarted.await(10, TimeUnit.SECONDS)); + assertTrue(subtask.disallowSubmittingSelf()); + if (drop) { + subtask.close(); + subtaskClosed = true; + } + supplyReleased.countDown(); + + assertFalse(execution.get(10, TimeUnit.SECONDS)); + assertFalse(subtask.executeOnce()); + verify(sink, never()).transfer(event); + verify(event, never()).decreaseReferenceCount(anyString(), anyBoolean()); + + if (drop) { + verify(event).clearReferenceCount(anyString()); + } else { + verifyZeroInteractions(sink); + subtask.allowSubmittingSelf(); + assertTrue(subtask.executeOnce()); + verify(source).supply(); + verify(sink).transfer(event); + verify(event).decreaseReferenceCount(anyString(), eq(true)); + } + } finally { + supplyReleased.countDown(); + worker.shutdownNow(); + assertTrue(worker.awaitTermination(10, TimeUnit.SECONDS)); + } + } +} From 60de6a52c81c6ce0319e4bff49938300e932fbb4 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 19 Oct 2026 15:15:54 +0800 Subject: [PATCH 5/8] [Pipe] Refresh pipe user id when credentials change --- .../config/TableConfigTaskVisitor.java | 46 ++++++ .../PipeAuthenticationIdentityTest.java | 131 ++++++++++++++++++ 2 files changed, 177 insertions(+) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/PipeAuthenticationIdentityTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java index 2af8eccc17388..f91969ae32ad3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java @@ -1370,6 +1370,15 @@ public static void checkAndEnrichSourceUser( .FAILED_TO_S_PIPE_S_IN_IOTDB_SOURCE_PASSWORD_MUST_BE_SET_WHEN_THE_USERNAME_IS_SPECIFIED, isAlter ? DataNodeQueryMessages.ALTER : DataNodeQueryMessages.CREATE, pipeName)); + } else { + updatePipeUserIdFromUsername( + replacedSourceAttributes, + sourceParameters, + PipeSourceConstant.SOURCE_IOTDB_USER_ID, + PipeSourceConstant.EXTRACTOR_IOTDB_USER_KEY, + PipeSourceConstant.SOURCE_IOTDB_USER_KEY, + PipeSourceConstant.EXTRACTOR_IOTDB_USERNAME_KEY, + PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY); } } @@ -1442,6 +1451,15 @@ public static void checkAndEnrichSinkUser( .FAILED_TO_S_PIPE_S_IN_WRITE_BACK_SINK_PASSWORD_MUST_BE_SET_WHEN_THE_USERNAME_IS, isAlter ? DataNodeQueryMessages.ALTER : DataNodeQueryMessages.CREATE, pipeName)); + } else { + updatePipeUserIdFromUsername( + connectorAttributes, + connectorParameters, + PipeSinkConstant.SINK_IOTDB_USER_ID, + PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY, + PipeSinkConstant.SINK_IOTDB_USER_KEY, + PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY, + PipeSinkConstant.SINK_IOTDB_USERNAME_KEY); } } @@ -1510,6 +1528,14 @@ public static void markSourceAuthenticationAsExplicitIfNecessary( PipeSourceConstant.SOURCE_IOTDB_PASSWORD_KEY)) { sourceAttributes.put( SystemConstant.SOURCE_AUTHENTICATION_INJECTED_KEY, Boolean.FALSE.toString()); + updatePipeUserIdFromUsername( + sourceAttributes, + sourceParameters, + PipeSourceConstant.SOURCE_IOTDB_USER_ID, + PipeSourceConstant.EXTRACTOR_IOTDB_USER_KEY, + PipeSourceConstant.SOURCE_IOTDB_USER_KEY, + PipeSourceConstant.EXTRACTOR_IOTDB_USERNAME_KEY, + PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY); } } @@ -1524,6 +1550,26 @@ public static void markSinkAuthenticationAsExplicitIfNecessary( PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY, PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY)) { sinkAttributes.put(SystemConstant.SINK_AUTHENTICATION_INJECTED_KEY, Boolean.FALSE.toString()); + updatePipeUserIdFromUsername( + sinkAttributes, + sinkParameters, + PipeSinkConstant.SINK_IOTDB_USER_ID, + PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY, + PipeSinkConstant.SINK_IOTDB_USER_KEY, + PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY, + PipeSinkConstant.SINK_IOTDB_USERNAME_KEY); + } + } + + private static void updatePipeUserIdFromUsername( + final Map attributes, + final PipeParameters parameters, + final String userIdAttributeKey, + final String... usernameAttributeKeys) { + final String username = parameters.getStringByKeys(usernameAttributeKeys); + if (username != null) { + attributes.put( + userIdAttributeKey, String.valueOf(AuthorityChecker.getUserId(username).orElse(-1L))); } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/PipeAuthenticationIdentityTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/PipeAuthenticationIdentityTest.java new file mode 100644 index 0000000000000..741dd0171ba26 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/PipeAuthenticationIdentityTest.java @@ -0,0 +1,131 @@ +/* + * 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.execution.config; + +import org.apache.iotdb.commons.audit.UserEntity; +import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; +import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; +import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; +import org.apache.iotdb.db.auth.AuthorityChecker; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.powermock.core.classloader.annotations.PowerMockIgnore; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; + +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.when; +import static org.powermock.api.mockito.PowerMockito.mockStatic; + +@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"}) +@RunWith(PowerMockRunner.class) +@PrepareForTest(AuthorityChecker.class) +public class PipeAuthenticationIdentityTest { + + @Test + public void testModifiedCredentialsReplaceInjectedSuperUserId() { + mockStatic(AuthorityChecker.class); + when(AuthorityChecker.getUserId("thulab")).thenReturn(Optional.of(10001L)); + + final Map sourceAttributes = new HashMap<>(); + TableConfigTaskVisitor.checkAndEnrichSourceUser( + "pipe", sourceAttributes, new UserEntity(0L, "root", "host"), false); + + final Map sourceChanges = new HashMap<>(); + sourceChanges.put("username", "thulab"); + sourceChanges.put("password", "password"); + TableConfigTaskVisitor.markSourceAuthenticationAsExplicitIfNecessary(sourceChanges); + new PipeParameters(sourceAttributes) + .addOrReplaceEquivalentAttributes(new PipeParameters(sourceChanges)); + + final Map sinkAttributes = new HashMap<>(); + sinkAttributes.put(PipeSinkConstant.SINK_KEY, "write-back-sink"); + TableConfigTaskVisitor.checkAndEnrichSinkUser( + "pipe", sinkAttributes, new UserEntity(0L, "root", "host"), false); + + final Map sinkChanges = new HashMap<>(); + sinkChanges.put(PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY, "thulab"); + sinkChanges.put(PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY, "password"); + TableConfigTaskVisitor.markSinkAuthenticationAsExplicitIfNecessary(sinkChanges); + new PipeParameters(sinkAttributes) + .addOrReplaceEquivalentAttributes(new PipeParameters(sinkChanges)); + + assertEquals("10001", sourceAttributes.get(PipeSourceConstant.SOURCE_IOTDB_USER_ID)); + assertEquals( + Boolean.FALSE.toString(), + sourceAttributes.get(SystemConstant.SOURCE_AUTHENTICATION_INJECTED_KEY)); + assertEquals("10001", sinkAttributes.get(PipeSinkConstant.SINK_IOTDB_USER_ID)); + assertEquals( + Boolean.FALSE.toString(), + sinkAttributes.get(SystemConstant.SINK_AUTHENTICATION_INJECTED_KEY)); + } + + @Test + public void testExplicitCredentialsUseTheirOwnUserIdOnCreateAndReplace() { + mockStatic(AuthorityChecker.class); + when(AuthorityChecker.getUserId("thulab")).thenReturn(Optional.of(10001L)); + + for (final boolean isAlter : new boolean[] {false, true}) { + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put(PipeSourceConstant.EXTRACTOR_IOTDB_USER_KEY, "thulab"); + sourceAttributes.put(PipeSourceConstant.EXTRACTOR_IOTDB_PASSWORD_KEY, "password"); + TableConfigTaskVisitor.checkAndEnrichSourceUser( + "pipe", sourceAttributes, new UserEntity(0L, "root", "host"), isAlter); + + final Map sinkAttributes = new HashMap<>(); + sinkAttributes.put(PipeSinkConstant.SINK_KEY, "write-back-sink"); + sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_USERNAME_KEY, "thulab"); + sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY, "password"); + TableConfigTaskVisitor.checkAndEnrichSinkUser( + "pipe", sinkAttributes, new UserEntity(0L, "root", "host"), isAlter); + + assertEquals("10001", sourceAttributes.get(PipeSourceConstant.SOURCE_IOTDB_USER_ID)); + assertEquals("10001", sinkAttributes.get(PipeSinkConstant.SINK_IOTDB_USER_ID)); + } + } + + @Test + public void testPasswordOnlyModificationPreservesExistingUserId() { + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put(PipeSourceConstant.SOURCE_IOTDB_USER_ID, "10001"); + final Map sourceChanges = new HashMap<>(); + sourceChanges.put("password", "new-password"); + TableConfigTaskVisitor.markSourceAuthenticationAsExplicitIfNecessary(sourceChanges); + new PipeParameters(sourceAttributes) + .addOrReplaceEquivalentAttributes(new PipeParameters(sourceChanges)); + + final Map sinkAttributes = new HashMap<>(); + sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_USER_ID, "10001"); + final Map sinkChanges = new HashMap<>(); + sinkChanges.put("password", "new-password"); + TableConfigTaskVisitor.markSinkAuthenticationAsExplicitIfNecessary(sinkChanges); + new PipeParameters(sinkAttributes) + .addOrReplaceEquivalentAttributes(new PipeParameters(sinkChanges)); + + assertEquals("10001", sourceAttributes.get(PipeSourceConstant.SOURCE_IOTDB_USER_ID)); + assertEquals("10001", sinkAttributes.get(PipeSinkConstant.SINK_IOTDB_USER_ID)); + } +} From 496c51a7fdc1ecf33de542c32adac4cf5b4e0b9f Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:50:02 +0800 Subject: [PATCH 6/8] [Pipe] Preserve TsFile privilege metadata across retries --- .../common/tsfile/PipeTsFileInsertionEvent.java | 7 +++---- .../event/PipeTsFileInsertionEventTest.java | 17 ++++++++++++++++- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 9690eed5d758a..9fc46f47e695d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -556,6 +556,9 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa holderMessage), e); } finally { + // Retries must repeat the privilege check, so release its metadata only with the event. + tableNames = null; + treeSchemaMap = null; if (Objects.nonNull(pipeName)) { try { PipeDataNodeSinglePipeMetrics.getInstance() @@ -932,10 +935,6 @@ else if (Objects.nonNull(treeSchemaMap)) { e.getMessage()); LOGGER.warn(errorMsg, e); throw new PipeException(errorMsg, e); - } finally { - // GC useless - tableNames = null; - treeSchemaMap = null; } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java index 5d5be1dbadee9..10f642527dcd9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java @@ -80,7 +80,8 @@ public void testAuthCheck() throws Exception { TsFileNameGenerator.generateNewTsFilePath( TestConstant.BASE_OUTPUT_PATH + IoTDBConstant.SEQUENCE_FOLDER_NAME, 1, 1, 1, 1)); try { - AuthorityChecker.setAccessControl(new TestAccessControl()); + final TestAccessControl accessControl = Mockito.spy(new TestAccessControl()); + AuthorityChecker.setAccessControl(accessControl); final TsFileResource resource = new TsFileResource(file); try (CompactionTestFileWriter writer = new CompactionTestFileWriter(resource)) { @@ -126,6 +127,18 @@ public void testAuthCheck() throws Exception { Long.MIN_VALUE, Long.MAX_VALUE); Assert.assertThrows(AccessDeniedException.class, tableEvent::throwIfNoPrivilege); + // A denied event must remain retryable until SELECT is granted. + Assert.assertThrows(AccessDeniedException.class, tableEvent::throwIfNoPrivilege); + Mockito.doReturn(true) + .when(accessControl) + .checkCanSelectFromTable4Pipe(Mockito.anyString(), Mockito.any(), Mockito.any()); + tableEvent.throwIfNoPrivilege(); + // Processing can fail after a successful check, so the next attempt must check again. + tableEvent.throwIfNoPrivilege(); + Mockito.doReturn(false) + .when(accessControl) + .checkCanSelectFromTable4Pipe(Mockito.anyString(), Mockito.any(), Mockito.any()); + Assert.assertThrows(AccessDeniedException.class, tableEvent::throwIfNoPrivilege); tableEvent.close(); final PipeTsFileInsertionEvent treeEvent = @@ -157,6 +170,8 @@ public void testAuthCheck() throws Exception { treeEvent.setTreeSchemaMap(Collections.singletonMap(deviceID, new String[] {"s0", "s1"})); Assert.assertThrows(AccessDeniedException.class, treeEvent::throwIfNoPrivilege); + // Retrying a realtime event must not fall back to the historical coarse filter. + Assert.assertThrows(AccessDeniedException.class, treeEvent::throwIfNoPrivilege); treeEvent.close(); } finally { From 8d12e8ea4a5ca24452e0d639fd74d2e3b04dd6a6 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 19 Oct 2026 19:35:21 +0800 Subject: [PATCH 7/8] [Fix] Retain LOAD payloads until V1 reports replication progress Treat the default IoTConsensus V1 WAL watermark as unconfirmed replication in both retention and the staged-directory cleaner. Keep immediate release for V2 and protocols without a V1 WAL. Cover commit payload readability, terminal task recovery, and cleanup before the first watermark. --- .../storageengine/load/LoadTaskRetention.java | 26 +++++----- .../storageengine/load/LoadTsFileCleaner.java | 24 ++++++---- .../load/LoadTsFileCleanerTest.java | 42 ++++++++++++++++ .../load/LoadTsFileManagerTest.java | 48 ++++++++++++++++--- 4 files changed, 111 insertions(+), 29 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTaskRetention.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTaskRetention.java index 828df844ed7a0..6197bbccc9895 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTaskRetention.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTaskRetention.java @@ -62,14 +62,7 @@ final class LoadTaskRetention { /** Checks whether the staged directory must be retained for follower WAL synchronization. */ boolean mustRetain(final long searchIndex) { - // IoTConsensusV2 replicates through pipe, so no replica reads these bytes back out of the WAL - // and there is no index to wait for: the staged directory can go as soon as the task ends. - if (LoadConsensusCompat.usesIoTConsensusV2()) { - return false; - } - final long safelyDeletedSearchIndex = currentSafelyDeletedSearchIndex(); - return safelyDeletedSearchIndex != ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX - && safelyDeletedSearchIndex < searchIndex; + return currentSafelyDeletedSearchIndex() < searchIndex; } /** Takes ownership of a staged directory, postponing deletion until WAL watermarks allow it. */ @@ -122,14 +115,14 @@ void release() { } final long safelyDeletedSearchIndex = currentSafelyDeletedSearchIndex(); - final long boundary = - safelyDeletedSearchIndex == ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX - ? Long.MAX_VALUE - : safelyDeletedSearchIndex; + // The default V1 watermark means no replica progress has been confirmed yet. + if (safelyDeletedSearchIndex == ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX) { + return; + } for (final Map.Entry entry : loadId2RetainedTaskDir.entrySet()) { final RetainedTaskDir retained = entry.getValue(); - if (retained.reachedSearchIndex() <= boundary + if (retained.reachedSearchIndex() <= safelyDeletedSearchIndex && loadId2RetainedTaskDir.remove(entry.getKey(), retained)) { LOGGER.info( StorageEngineMessages @@ -165,10 +158,15 @@ private void ensureWalListenerRegistered() { } private long currentSafelyDeletedSearchIndex() { + // V2 replicates through pipe. Other protocols without a V1 WAL (for example Ratis) do not + // expand staged references from this node's WAL either, so neither has a watermark to wait for. + if (LoadConsensusCompat.usesIoTConsensusV2()) { + return Long.MAX_VALUE; + } return dataRegion .getWALNode() .map(IWALNode::getSafelyDeletedSearchIndex) - .orElse(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); + .orElse(Long.MAX_VALUE); } // ------------------------------------------------------------------------- diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleaner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleaner.java index ee440de7790f7..93a537c1e4aca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleaner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleaner.java @@ -350,14 +350,22 @@ private static boolean isComplete(final File taskDir) { } private static boolean replicasReached(final DataRegion dataRegion, final long searchIndex) { - final long safelyDeletedSearchIndex = - dataRegion - .getWALNode() - .map(wal -> wal.getSafelyDeletedSearchIndex()) - .orElse(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); - - return safelyDeletedSearchIndex == ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX - || safelyDeletedSearchIndex >= searchIndex; + // IoTConsensusV2 and consensus implementations without a WAL replicate the staged payload + // directly, so no follower can read it back from this node's WAL. IoTConsensus V1, however, + // must report a real watermark before its staged references are safe to delete. + if (LoadConsensusCompat.usesIoTConsensusV2()) { + return true; + } + return dataRegion + .getWALNode() + .map( + wal -> { + final long safelyDeletedSearchIndex = wal.getSafelyDeletedSearchIndex(); + return safelyDeletedSearchIndex + != ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX + && safelyDeletedSearchIndex >= searchIndex; + }) + .orElse(true); } // ------------------------------------------------------------------------- diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleanerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleanerTest.java index ca25a96ad4bbe..01482f2dfe045 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleanerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleanerTest.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.conf.ConfigurationFileUtils; import org.apache.iotdb.commons.conf.TrimProperties; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusNode; @@ -186,6 +187,47 @@ public void testAnIdleTaskIsReclaimedOnlyOnceItsPiecesAreReplicated() throws Exc } } + @Test + public void testAFinishedTaskWaitsForTheFirstWatermark() throws Exception { + final String loadId = "unreported-watermark-task"; + final File taskDir = new File(tempDir, loadId); + Assert.assertTrue(taskDir.mkdirs()); + stageOneChunk(taskDir, 5L); + final LoadTsFileCleaner cleaner = LoadTsFileCleaner.getInstance(); + cleaner.register(loadId, dataRegion, taskDir, 7L, LoadTsFileConsensusOp.ABORT); + + walWatermark.set(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); + cleaner.sweepRegisteredTasks(); + Assert.assertTrue("an unreported watermark does not confirm replication", taskDir.exists()); + + walWatermark.set(5L); + cleaner.sweepRegisteredTasks(); + Assert.assertTrue("the terminal command is not replicated yet", taskDir.exists()); + + walWatermark.set(7L); + cleaner.sweepRegisteredTasks(); + Assert.assertFalse("the terminal command is replicated", taskDir.exists()); + } + + @Test + public void testIoTConsensusV2DoesNotWaitForAWALWatermark() throws Exception { + final String loadId = "v2-finished-task"; + final File taskDir = new File(tempDir, loadId); + Assert.assertTrue(taskDir.mkdirs()); + stageOneChunk(taskDir, 5L); + final LoadTsFileCleaner cleaner = LoadTsFileCleaner.getInstance(); + cleaner.register(loadId, dataRegion, taskDir, 7L, LoadTsFileConsensusOp.ABORT); + walWatermark.set(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); + + LoadConsensusCompat.setIoTConsensusV2ForTest(Boolean.TRUE); + try { + cleaner.sweepRegisteredTasks(); + Assert.assertFalse("V2 does not expand staged references from the WAL", taskDir.exists()); + } finally { + LoadConsensusCompat.setIoTConsensusV2ForTest(null); + } + } + @Test public void testAFinishedCommittedTaskIsReleasedOnceItsStagedFilesAreComplete() throws Exception { final String loadId = "finished-task"; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java index c98b3afa19583..360fa83c6afbd 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java @@ -21,6 +21,7 @@ import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadFileException; @@ -634,10 +635,8 @@ private static NonAlignedChunkData[] twoChunksOfTheSamePartition() { } @Test - public void testAbortedConsensusTaskReleasesStagedFilesWhenNoWatermarkIsReported() - throws Exception { - // Without a consensus watermark (single replica) nothing can read the staged bytes back, so the - // directory of a finished task is deleted right away. + public void testAbortedConsensusTaskReleasesStagedFilesWithoutAWAL() throws Exception { + // Protocols without a V1 WAL do not expand staged references from the WAL. final String loadId = "aborted-load"; final LoadTsFileManager manager = new LoadTsFileManager(dataRegion); final List refs = @@ -660,8 +659,8 @@ public void testFinishedConsensusTaskKeepsStagedFilesUntilTheWatermarkAdvances() try (final WALNode walNode = new WALNode("load-retention-test", new File(tempDir, "wal").getAbsolutePath())) { Mockito.when(dataRegion.getWALNode()).thenReturn(Optional.of(walNode)); - // The watermark lags behind the command, so a follower may still need the staged bytes. - walNode.setSafelyDeletedSearchIndex(5L); + // Until V1 reports its first watermark, a follower may still need the staged bytes. + walNode.setSafelyDeletedSearchIndex(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); final LoadTsFileManager manager = new LoadTsFileManager(dataRegion); final String loadId = "retained-load"; @@ -682,6 +681,10 @@ public void testFinishedConsensusTaskKeepsStagedFilesUntilTheWatermarkAdvances() assertTrue(progress.isFile()); assertNotNull(LoadTsFileProgress.readTerminal(progress)); + // A reported watermark that still lags behind the command must keep the staged bytes too. + walNode.setSafelyDeletedSearchIndex(5L); + assertTrue(staged.isFile()); + // Once every follower got past the command, the directory is released. walNode.setSafelyDeletedSearchIndex(10L); assertFalse(staged.exists()); @@ -708,6 +711,27 @@ public void testStagedDirectoryOfFinishedTaskIsDeletedOnRecovery() throws Except assertFalse(manager.prepare(prepareNode("finished-load"), Collections.emptyMap())); } + @Test + public void testRecoveredFinishedTaskWaitsForTheFirstWatermark() throws Exception { + useWatermarkWALNode(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); + final File taskDir = new File(tempDir, "root.load_manager_test-0/recovered-finished-load"); + assertTrue(taskDir.mkdirs()); + final File stagedFile = new File(taskDir, "123.tsfile"); + Files.write(stagedFile.toPath(), new byte[] {1, 2, 3}); + new LoadTsFileProgress(stagedFile).recordTerminal(LoadTsFileConsensusOp.ABORT, 7L); + + final LoadTsFileManager manager = new LoadTsFileManager(dataRegion); + try { + assertTrue("a restart must keep bytes until replication is confirmed", stagedFile.isFile()); + assertFalse(manager.prepare(prepareNode(taskDir.getName()), Collections.emptyMap())); + + advanceWatermark(7L); + assertFalse(taskDir.exists()); + } finally { + manager.stop(); + } + } + @Test public void testIoTConsensusV2WaitsForNoIndexBeforeReleasing() { LoadConsensusCompat.setIoTConsensusV2ForTest(Boolean.TRUE); @@ -778,7 +802,17 @@ public void testStagedDirectoryWithoutTerminalRecordIsKeptOnRecovery() throws Ex public void testCommitKeepsTheStagedBytesWhileAReplicaMayStillNeedThem() throws Exception { // The watermark lags behind the COMMIT command, so a replica that catches up from this node's // WAL still has to expand the pieces of the task from the staging directory. - useWatermarkWALNode(5L); + assertCommitKeepsTheStagedBytes(5L); + } + + @Test + public void testCommitKeepsTheStagedBytesBeforeTheFirstWatermark() throws Exception { + assertCommitKeepsTheStagedBytes(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); + } + + private void assertCommitKeepsTheStagedBytes(final long safelyDeletedSearchIndex) + throws Exception { + useWatermarkWALNode(safelyDeletedSearchIndex); emulateStagedFileImport(); final String loadId = "committed-load"; From df0d45733e5f4a8d84510ace56dd2c8a46f926bb Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 20 Oct 2026 12:06:49 +0800 Subject: [PATCH 8/8] Fix IoTConsensusV2 progress for split LOAD TsFiles (cherry picked from commit 05c3d81ff75c5bc8f512d525dddadeca9b553850) --- .../load/TsFileWriterManager.java | 11 ++- .../load/LoadTsFileManagerTest.java | 70 +++++++++++++++++++ 2 files changed, 80 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java index fc8f38b850288..d3d9efb1345a5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java @@ -31,6 +31,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadFileException; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusOp; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; @@ -532,7 +533,7 @@ void prepare( /** Formally imports sealed staged TsFiles into the target storage engine data regions. */ void loadAll(final boolean isGeneratedByPipe, final boolean deleteStagedSource) - throws LoadFileException { + throws IOException, LoadFileException { for (final Map.Entry entry : timePartition2Resource.entrySet()) { final DataRegion targetDataRegion = dataRegion; @@ -547,6 +548,14 @@ void loadAll(final boolean isGeneratedByPipe, final boolean deleteStagedSource) uuidDirName(tsFileResource.getTsFile()))); } + if (LoadConsensusCompat.usesIoTConsensusV2()) { + // V2 only replicates files carrying this node's progress. The coordinator's index may + // belong to another node and precede writes between PREPARE and COMMIT. Assign progress + // at import, and persist it so recovery keeps the file eligible for replication. + PipeDataNodeAgent.runtime().assignProgressIndexForTsFileLoad(tsFileResource); + tsFileResource.serialize(); + } + targetDataRegion.loadNewTsFile( tsFileResource, deleteStagedSource, isGeneratedByPipe, false, Optional.empty()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java index 360fa83c6afbd..a3d2473316760 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java @@ -20,16 +20,22 @@ package org.apache.iotdb.db.storageengine.load; import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; +import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadFileException; +import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.pipe.processor.iotconsensusv2.IoTConsensusV2Processor; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusOp; import org.apache.iotdb.db.queryengine.plan.scheduler.load.ChunkOffsetCalculator; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; import org.apache.iotdb.db.storageengine.dataregion.wal.node.IWALNode; import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode; @@ -798,6 +804,70 @@ public void testStagedDirectoryWithoutTerminalRecordIsKeptOnRecovery() throws Ex assertFalse(manager.prepare(prepareNode("in-progress-load"), Collections.emptyMap())); } + @Test + public void testIoTConsensusV2CommitUsesReceivingNodeProgress() throws Exception { + assertCommittedProgress(true, config.getDataNodeId() + 1); + } + + @Test + public void testIoTConsensusV2CommitAdvancesPreparedProgress() throws Exception { + assertCommittedProgress(true, config.getDataNodeId()); + } + + @Test + public void testOtherConsensusCommitPreservesCoordinatorProgress() throws Exception { + assertCommittedProgress(false, config.getDataNodeId() + 1); + } + + private void assertCommittedProgress(final boolean useIoTConsensusV2, final int coordinatorNodeId) + throws Exception { + LoadConsensusCompat.setIoTConsensusV2ForTest(useIoTConsensusV2); + final String loadId = "load-progress"; + final LoadTsFileManager manager = new LoadTsFileManager(dataRegion); + try { + emulateStagedFileImport(); + final LoadTsFileConsensusNode piece = stagedPiece(loadId, laidOutChunk()); + manager.writePiece(piece); + final File staged = new File(tempDir, piece.getPieceRefs().get(0).getRelativePath()); + final RecoverProgressIndex preparedLocalProgress = + PipeDataNodeAgent.runtime().getNextProgressIndexForTsFileLoad(); + final ProgressIndex coordinatorProgress = + new RecoverProgressIndex( + coordinatorNodeId, + preparedLocalProgress.getDataNodeId2LocalIndex().get(config.getDataNodeId())); + assertTrue( + manager.prepare( + prepareNode(loadId), + Collections.singletonMap(new TTimePartitionSlot(0L), coordinatorProgress))); + + // A local write after PREPARE must precede the progress assigned when the file is imported. + final ProgressIndex interveningWrite = + PipeDataNodeAgent.runtime().getNextProgressIndexForTsFileLoad(); + final LoadTsFileConsensusNode commit = commitNode(loadId); + assertTrue(manager.loadAll(commit, Collections.emptyMap())); + + final TsFileResource imported = + new TsFileResource(new File(new File(tempDir, "data"), staged.getName())); + imported.deserialize(); + imported.setStatus(TsFileResourceStatus.NORMAL); + final ProgressIndex persistedProgress = imported.getMaxProgressIndexAfterClose(); + final PipeTsFileInsertionEvent event = Mockito.mock(PipeTsFileInsertionEvent.class); + Mockito.when(event.forceGetProgressIndex()).thenReturn(persistedProgress); + assertEquals( + "A V2 LOAD must remain eligible for consensus replication after resource recovery", + useIoTConsensusV2, + IoTConsensusV2Processor.isShouldReplicate(event)); + if (useIoTConsensusV2) { + assertTrue(persistedProgress.isAfter(interveningWrite)); + } else { + assertEquals(coordinatorProgress, persistedProgress); + } + } finally { + manager.stop(); + LoadConsensusCompat.setIoTConsensusV2ForTest(null); + } + } + @Test public void testCommitKeepsTheStagedBytesWhileAReplicaMayStillNeedThem() throws Exception { // The watermark lags behind the COMMIT command, so a replica that catches up from this node's