You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/02/04 09:32:36 UTC
(iotdb) branch rc/2.0.6 updated (fbb4c351b32 -> 155789d75b2)
This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a change to branch rc/2.0.6
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from fbb4c351b32 [maven-release-plugin] prepare for next development iteration
new 533f0af6404 fix
new 155789d75b2 File sink
The 2 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.
Summary of changes:
.../dataregion/PipeDataRegionSinkConstructor.java | 4 +
.../scan/TsFileInsertionEventScanParser.java | 12 +-
.../iotdb/db/pipe/sink/protocol/ToFileSink.java | 369 +++++++++++++++++++++
.../agent/plugin/builtin/BuiltinPipePlugin.java | 2 +
4 files changed, 385 insertions(+), 2 deletions(-)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/ToFileSink.java
(iotdb) 01/02: fix
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch rc/2.0.6
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 533f0af6404d4c07cbd38d2d804f8c5e344df806
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Wed Feb 4 09:44:36 2026 +0800
fix
---
.../tsfile/parser/scan/TsFileInsertionEventScanParser.java | 12 ++++++++++--
1 file changed, 10 insertions(+), 2 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
index 13d95eadeb1..16a4dd168b1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
@@ -444,7 +444,11 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
tsFileSequenceReader.readChunk(-1, chunkHeader.getDataSize())));
currentIsAligned = false;
currentMeasurements.add(
- new MeasurementSchema(chunkHeader.getMeasurementID(), chunkHeader.getDataType()));
+ new MeasurementSchema(
+ chunkHeader.getMeasurementID(),
+ chunkHeader.getDataType(),
+ chunkHeader.getEncodingType(),
+ chunkHeader.getCompressionType()));
return;
case MetaMarker.VALUE_CHUNK_HEADER:
case MetaMarker.ONLY_ONE_PAGE_VALUE_CHUNK_HEADER:
@@ -506,7 +510,11 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
new Chunk(
chunkHeader, tsFileSequenceReader.readChunk(-1, chunkHeader.getDataSize())));
currentMeasurements.add(
- new MeasurementSchema(chunkHeader.getMeasurementID(), chunkHeader.getDataType()));
+ new MeasurementSchema(
+ chunkHeader.getMeasurementID(),
+ chunkHeader.getDataType(),
+ chunkHeader.getEncodingType(),
+ chunkHeader.getCompressionType()));
break;
case MetaMarker.CHUNK_GROUP_HEADER:
// Return before "currentDevice" changes
(iotdb) 02/02: File sink
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch rc/2.0.6
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 155789d75b211ce72120c72060b4dadba9363aa8
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Wed Feb 4 17:26:20 2026 +0800
File sink
---
.../dataregion/PipeDataRegionSinkConstructor.java | 4 +
.../iotdb/db/pipe/sink/protocol/ToFileSink.java | 369 +++++++++++++++++++++
.../agent/plugin/builtin/BuiltinPipePlugin.java | 2 +
3 files changed, 375 insertions(+)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java
index 09773d0cad5..58d72295ef1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.donothing.DoNothingSink;
import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeSinkConstructor;
import org.apache.iotdb.commons.pipe.agent.plugin.meta.DataNodePipePluginMetaKeeper;
+import org.apache.iotdb.db.pipe.sink.protocol.ToFileSink;
import org.apache.iotdb.db.pipe.sink.protocol.legacy.IoTDBLegacyPipeSink;
import org.apache.iotdb.db.pipe.sink.protocol.pipeconsensus.PipeConsensusAsyncSink;
import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
@@ -62,6 +63,8 @@ class PipeDataRegionSinkConstructor extends PipeSinkConstructor {
BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingSink::new);
pluginConstructors.put(
BuiltinPipePlugin.WRITE_BACK_CONNECTOR.getPipePluginName(), WriteBackSink::new);
+ pluginConstructors.put(
+ BuiltinPipePlugin.TO_FILE_CONNECTOR.getPipePluginName(), ToFileSink::new);
pluginConstructors.put(
BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName(), IoTDBDataRegionAsyncSink::new);
@@ -85,5 +88,6 @@ class PipeDataRegionSinkConstructor extends PipeSinkConstructor {
pluginConstructors.put(
BuiltinPipePlugin.PIPE_CONSENSUS_ASYNC_SINK.getPipePluginName(),
PipeConsensusAsyncSink::new);
+ pluginConstructors.put(BuiltinPipePlugin.TO_FILE_SINK.getPipePluginName(), ToFileSink::new);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/ToFileSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/ToFileSink.java
new file mode 100644
index 00000000000..f5b2d4111a4
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/ToFileSink.java
@@ -0,0 +1,369 @@
+/*
+ * IoTDB Pipe框架自定义Sink插件
+ *
+ * 功能：将 Pipe 下发的数据以 TsFile 形式写入本地指定目录，按库名 / 导出日期创建子目录。
+ * TabletInsertionEvent 直接按 库(db) + 导出日期 写成 tsfile（同一库下多 device 共用一个文件）；
+ * TsFileInsertionEvent 使用 PipeConnector 默认逻辑转为 TabletInsertionEvent 后同样写入，
+ * 不再做 TsFile 复制或拆分，避免大文件重写带来的 GC 压力。
+ *
+ * 使用方法示例：
+ * CREATE PIPE pipe1
+ * WITH CONNECTOR (
+ * 'connector'='ToFileSink',
+ * 'sink.directory'='/path/to/save/tsfiles'
+ * );
+ */
+
+package org.apache.iotdb.db.pipe.sink.protocol;
+
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventTsFileBatch;
+import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTransferBatchReqBuilder;
+import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
+import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
+import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+
+import org.apache.tsfile.exception.write.WriteProcessException;
+import org.apache.tsfile.utils.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * 将 Pipe 数据以 TsFile 形式写入本地目录的 Sink。
+ *
+ * <p>默认使用 {@link PipeTabletEventTsFileBatch} 批处理：仅支持 PipeInsertNodeTabletInsertionEvent /
+ * PipeRawTabletInsertionEvent，按批 flush 到 sink.directory，按库名/日期子目录布局。心跳事件用于触发落盘，避免最后几条留在内存。
+ */
+@TreeModel
+@TableModel
+public class ToFileSink implements PipeConnector {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ToFileSink.class);
+
+ // 配置参数键
+ private static final String SINK_DIRECTORY_KEY = "sink.directory";
+ private static final String CONNECTOR_DIRECTORY_KEY = "connector.directory";
+ private static final String SINK_CREATE_SUBDIRS_KEY = "sink.create-subdirs";
+
+ /** 批处理：最大延迟（毫秒）触发 flush */
+ private static final String SINK_BATCH_DELAY_MS_KEY = "sink.batch-delay-ms";
+
+ private static final String CONNECTOR_BATCH_DELAY_MS_KEY = "connector.batch-delay-ms";
+
+ /** 批处理：最大缓冲字节数触发 flush */
+ private static final String SINK_BATCH_SIZE_BYTES_KEY = "sink.batch-size-bytes";
+
+ private static final String CONNECTOR_BATCH_SIZE_BYTES_KEY = "connector.batch-size-bytes";
+
+ /** 默认批延迟偏小，避免多 region 时 Pipe tablet 内存超限（flush 越早越早释放内存） */
+ private static final int DEFAULT_BATCH_DELAY_MS = 500;
+
+ private static final long DEFAULT_BATCH_SIZE_BYTES = 20L * 1024 * 1024;
+
+ // 配置参数
+ private String targetDirectory;
+ private boolean createSubdirs = false;
+ private String pipeName;
+
+ /** 实例唯一标识，用于 sink.parallel.tasks>1 时避免多实例写同一目录文件名冲突 */
+ private String instanceId;
+
+ /** TsFile 批处理器（默认唯一写路径） */
+ private PipeTabletEventTsFileBatch tsFileBatch;
+
+ private final Object batchFlushLock = new Object();
+
+ private static final DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("yyyyMMdd");
+
+ /** 文件名时间戳：精确到毫秒，避免同一秒内多次 flush 覆盖同一文件 */
+ private static final DateTimeFormatter DATETIME_FORMAT =
+ DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS");
+
+ /** 实例内 flush 序号，保证同一毫秒内多次 flush 也不重名 */
+ private final AtomicLong flushSequence = new AtomicLong(0);
+
+ @Override
+ public void validate(PipeParameterValidator validator) throws Exception {
+ final PipeParameters parameters = validator.getParameters();
+
+ // 验证目录参数
+ final String dir =
+ parameters.getStringOrDefault(
+ Arrays.asList(SINK_DIRECTORY_KEY, CONNECTOR_DIRECTORY_KEY), null);
+ validator.validate(
+ arg -> {
+ final String directory = (String) arg;
+ return directory != null && !directory.trim().isEmpty();
+ },
+ String.format(
+ "The parameter %s or %s must be specified and not empty",
+ SINK_DIRECTORY_KEY, CONNECTOR_DIRECTORY_KEY),
+ dir);
+
+ // 验证目录路径
+ if (dir != null) {
+ final Path path = Paths.get(dir);
+ if (path.toFile().exists() && !path.toFile().isDirectory()) {
+ throw new PipeException(String.format("The path %s exists but is not a directory", dir));
+ }
+ }
+ }
+
+ @Override
+ public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration)
+ throws Exception {
+ // 获取目标目录
+ targetDirectory =
+ parameters.getStringOrDefault(
+ Arrays.asList(SINK_DIRECTORY_KEY, CONNECTOR_DIRECTORY_KEY), null);
+
+ if (targetDirectory == null || targetDirectory.trim().isEmpty()) {
+ throw new PipeException(
+ String.format(
+ "The parameter %s or %s must be specified",
+ SINK_DIRECTORY_KEY, CONNECTOR_DIRECTORY_KEY));
+ }
+
+ // 规范化路径
+ targetDirectory = Paths.get(targetDirectory).toAbsolutePath().normalize().toString();
+
+ createSubdirs =
+ parameters.getBooleanOrDefault(
+ Arrays.asList(SINK_CREATE_SUBDIRS_KEY, "connector.create-subdirs"), false);
+
+ // 获取Pipe信息
+ pipeName = configuration.getRuntimeEnvironment().getPipeName();
+ // 实例唯一标识，sink.parallel.tasks>1 时避免多实例写同一目录文件名冲突
+ instanceId =
+ Integer.toHexString(System.identityHashCode(this))
+ + "_"
+ + Long.toHexString(System.nanoTime() % 0xFFFF);
+
+ int batchDelayMs =
+ parameters.getIntOrDefault(
+ Arrays.asList(SINK_BATCH_DELAY_MS_KEY, CONNECTOR_BATCH_DELAY_MS_KEY),
+ DEFAULT_BATCH_DELAY_MS);
+ long batchSizeBytes =
+ parameters.getLongOrDefault(
+ Arrays.asList(SINK_BATCH_SIZE_BYTES_KEY, CONNECTOR_BATCH_SIZE_BYTES_KEY),
+ DEFAULT_BATCH_SIZE_BYTES);
+ tsFileBatch = new PipeTabletEventTsFileBatch(batchDelayMs, batchSizeBytes);
+
+ LOGGER.info(
+ "ToFileSink customized: targetDirectory={}, createSubdirs={}, pipeName={}, instanceId={}, batchDelayMs={}, batchSizeBytes={}",
+ targetDirectory,
+ createSubdirs,
+ pipeName,
+ instanceId,
+ batchDelayMs,
+ batchSizeBytes);
+ }
+
+ @Override
+ public void handshake() throws Exception {
+ // 创建目标目录
+ final Path targetPath = Paths.get(targetDirectory);
+ if (!Files.exists(targetPath)) {
+ Files.createDirectories(targetPath);
+ LOGGER.info("Created target directory: {}", targetDirectory);
+ }
+
+ // 验证目录可写
+ if (!Files.isWritable(targetPath)) {
+ throw new PipeException(
+ String.format("Target directory is not writable: %s", targetDirectory));
+ }
+
+ LOGGER.info("ToFileSink handshake completed: {}", targetDirectory);
+ }
+
+ @Override
+ public void heartbeat() throws Exception {
+ // 检查目标目录是否仍然存在且可写
+ final Path targetPath = Paths.get(targetDirectory);
+ if (!Files.exists(targetPath)) {
+ throw new PipeConnectionException(
+ String.format("Target directory does not exist: %s", targetDirectory));
+ }
+ if (!Files.isWritable(targetPath)) {
+ throw new PipeConnectionException(
+ String.format("Target directory is not writable: %s", targetDirectory));
+ }
+ }
+
+ /**
+ * 正确流程（按开发者约定）： 1. constructBatch：将事件放入批内存（此处通过 onEvent 实现）； 2. shouldEmit：判断内存大小或批时间是否超过预期，超过则返回
+ * true； 3. 若 shouldEmit 为 true，则手动调用 sealTsFiles 生成文件； 4. 若生成/写入文件成功，则调用 onSuccess。
+ */
+ @Override
+ public void transfer(TabletInsertionEvent tabletInsertionEvent) throws Exception {
+ if (!(tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent)
+ && !(tabletInsertionEvent instanceof PipeRawTabletInsertionEvent)) {
+ LOGGER.warn(
+ "ToFileSink only supports PipeInsertNodeTabletInsertionEvent and PipeRawTabletInsertionEvent, ignore {}",
+ tabletInsertionEvent.getClass().getSimpleName());
+ return;
+ }
+ try {
+ // 1. constructBatch：将事件放入内存（PipeTabletEventTsFileBatch 通过 onEvent 实现）
+ tsFileBatch.onEvent(tabletInsertionEvent);
+ // 2. shouldEmit：判断内存大小或批时间是否超过预期
+ if (tsFileBatch.shouldEmit()) {
+ // 3. 手动调用 sealTsFiles 生成文件，成功则在 flush 内调用 onSuccess
+ flushBatchToTargetDirectory();
+ }
+ } catch (WALPipeException | IOException e) {
+ throw new PipeException("ToFileSink: TsFile batch onEvent failed", e);
+ }
+ }
+
+ /**
+ * 收到 TsFile 事件时先 flush 当前批再解析 TsFile，避免批内占用的 tablet 内存不释放、 导致 toTabletInsertionEvents() 在
+ * waitForResourceEnough4Parsing 里等不到内存而超时。
+ */
+ @Override
+ public void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception {
+ if (tsFileBatch != null && !tsFileBatch.isEmpty()) {
+ try {
+ flushBatchToTargetDirectory();
+ } catch (IOException | WriteProcessException e) {
+ throw new PipeException("ToFileSink: flush before TsFile parse failed", e);
+ }
+ }
+ try {
+ for (final TabletInsertionEvent tabletInsertionEvent :
+ tsFileInsertionEvent.toTabletInsertionEvents()) {
+ transfer(tabletInsertionEvent);
+ }
+ } finally {
+ tsFileInsertionEvent.close();
+ }
+ }
+
+ /** 心跳等事件：若批内已有数据且 shouldEmit 为 true，则 sealTsFiles 并复制，成功则 onSuccess。 */
+ @Override
+ public void transfer(Event event) throws Exception {
+ if (tsFileBatch != null && !tsFileBatch.isEmpty() && tsFileBatch.shouldEmit()) {
+ try {
+ flushBatchToTargetDirectory();
+ } catch (IOException | WriteProcessException e) {
+ throw new PipeException("ToFileSink: heartbeat-triggered batch flush failed", e);
+ }
+ }
+ LOGGER.debug("ToFileSink received event: {}", event.getClass().getSimpleName());
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (tsFileBatch != null) {
+ synchronized (batchFlushLock) {
+ try {
+ if (!tsFileBatch.isEmpty()) {
+ flushBatchToTargetDirectory();
+ }
+ } catch (Exception e) {
+ LOGGER.warn(
+ "ToFileSink: flush remaining TsFile batch on close failed: {}", e.getMessage());
+ }
+ try {
+ tsFileBatch.close();
+ } catch (Exception e) {
+ LOGGER.warn("ToFileSink: close TsFile batch failed: {}", e.getMessage());
+ }
+ tsFileBatch = null;
+ }
+ }
+
+ LOGGER.info("ToFileSink closed: {}", targetDirectory);
+ }
+
+ /**
+ * 按约定：shouldEmit 为 true 时调用本方法。 内部执行 sealTsFiles 生成文件，将 TsFile 复制到目标目录（按 db/日期 布局）；
+ * 仅当生成/写入文件成功后调用 onSuccess。
+ */
+ private void flushBatchToTargetDirectory() throws IOException, WriteProcessException {
+ if (tsFileBatch == null || tsFileBatch.isEmpty()) {
+ return;
+ }
+ synchronized (batchFlushLock) {
+ // 3. 手动调用 sealTsFiles 生成文件
+ List<Pair<String, File>> dbTsFilePairs = tsFileBatch.sealTsFiles();
+ if (dbTsFilePairs.isEmpty()) {
+ tsFileBatch.decreaseEventsReferenceCount(
+ PipeTransferBatchReqBuilder.class.getName(), false);
+ // 无文件生成也视为本批处理完成，需 onSuccess 清空批状态
+ tsFileBatch.onSuccess();
+ return;
+ }
+ final String date = LocalDate.now(ZoneId.systemDefault()).format(DATE_FORMAT);
+ final String createTime =
+ Instant.now().atZone(ZoneId.systemDefault()).format(DATETIME_FORMAT);
+ final long seq = flushSequence.getAndIncrement();
+ int index = 0;
+ for (Pair<String, File> pair : dbTsFilePairs) {
+ String dbName = pair.left;
+ File srcFile = pair.right;
+ if (dbName == null) {
+ dbName = "";
+ }
+ Path basePath = Paths.get(targetDirectory);
+ if (createSubdirs && !dbName.isEmpty()) {
+ basePath = basePath.resolve(dbName.replace('.', '_').replace(' ', '_'));
+ }
+ Path dateDir = basePath.resolve(date);
+ Files.createDirectories(dateDir);
+ String safeDbName =
+ dbName.replace('.', '_').replace(' ', '_').replace('/', '_').replace(':', '_');
+ // 使用 毫秒时间戳 + 序号，避免同一秒/同一毫秒内多次 flush 覆盖
+ String fileName =
+ instanceId == null
+ ? String.format("tablet-%s-%s-%d-%d.tsfile", safeDbName, createTime, seq, index)
+ : String.format(
+ "tablet-%s-%s-%d-%d-%s.tsfile", safeDbName, createTime, seq, index, instanceId);
+ Path destPath = dateDir.resolve(fileName);
+ Files.copy(srcFile.toPath(), destPath, StandardCopyOption.REPLACE_EXISTING);
+ try {
+ if (!srcFile.delete()) {
+ LOGGER.warn("ToFileSink: could not delete batch temp file {}", srcFile);
+ }
+ } catch (Exception e) {
+ LOGGER.warn(
+ "ToFileSink: failed to delete batch temp file {}: {}", srcFile, e.getMessage());
+ }
+ LOGGER.debug("ToFileSink: batch TsFile copied to {}", destPath);
+ index++;
+ }
+ // 释放本批事件的引用，使 tablet 内存立即归还 PipeMemoryManager
+ tsFileBatch.decreaseEventsReferenceCount(PipeTransferBatchReqBuilder.class.getName(), false);
+ // 4. 生成文件并复制成功后调用 onSuccess
+ tsFileBatch.onSuccess();
+ LOGGER.info(
+ "ToFileSink: flushed TsFile batch, {} file(s) copied to targetDirectory",
+ dbTsFilePairs.size());
+ }
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java
index daab48bbe23..d0087160e3e 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java
@@ -69,6 +69,7 @@ public enum BuiltinPipePlugin {
WEBSOCKET_CONNECTOR("websocket-connector", WebSocketSink.class),
WRITE_BACK_CONNECTOR("write-back-connector", WriteBackSink.class),
+ TO_FILE_CONNECTOR("to-file-connector", WriteBackSink.class),
DO_NOTHING_SINK("do-nothing-sink", DoNothingSink.class),
IOTDB_THRIFT_SINK("iotdb-thrift-sink", IoTDBThriftSink.class),
@@ -80,6 +81,7 @@ public enum BuiltinPipePlugin {
WRITE_BACK_SINK("write-back-sink", WriteBackSink.class),
SUBSCRIPTION_SINK("subscription-sink", DoNothingSink.class),
PIPE_CONSENSUS_ASYNC_SINK("pipe-consensus-async-sink", PipeConsensusAsyncSink.class),
+ TO_FILE_SINK("to-file-sink", WriteBackSink.class),
;
private final String pipePluginName;