change
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileRemotely.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileRemotely.java index aca33ba..eed5a7f 100644 --- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileRemotely.java +++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileRemotely.java
@@ -30,10 +30,10 @@ import org.apache.iotdb.commons.pipe.sink.payload.thrift.response.PipeTransferFilePieceResp; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferMultiFilePieceReq; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferMultiFileSealReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceWithModReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealWithModReq; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.pipe.api.exception.PipeConnectionException; import org.apache.iotdb.pipe.api.exception.PipeException; @@ -205,7 +205,7 @@ transferFilePieces(tsFile, true); req = - PipeTransferTsFileSealWithModReq.toTPipeTransferReq( + PipeTransferMultiFileSealReq.toTPipeTransferReq( modFile.getName(), modFile.length(), tsFile.getName(), tsFile.length()); } else { transferFilePieces(tsFile, false); @@ -286,7 +286,7 @@ private PipeTransferFilePieceReq getTransferMultiFilePieceReq( final String fileName, final long position, final byte[] payLoad) throws IOException { - return PipeTransferTsFilePieceWithModReq.toTPipeTransferReq(fileName, position, payLoad); + return PipeTransferMultiFilePieceReq.toTPipeTransferReq(fileName, position, payLoad); } private PipeTransferFilePieceReq getTransferSingleFilePieceReq(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java index 8965fe4..d629d72 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -61,6 +61,8 @@ import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferMultiFilePieceReq; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferMultiFileSealReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferPlanNodeReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferSchemaSnapshotPieceReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferSchemaSnapshotSealReq; @@ -73,9 +75,7 @@ import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletRawReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletRawReqV2; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceWithModReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealWithModReq; import org.apache.iotdb.db.protocol.basic.BasicOpenSessionResp; import org.apache.iotdb.db.protocol.session.IClientSession; import org.apache.iotdb.db.protocol.session.SessionManager; @@ -347,11 +347,11 @@ .recordTransferTsFileSealTimer(System.nanoTime() - startTime); } } - case TRANSFER_TS_FILE_PIECE_WITH_MOD: + case TRANSFER_MULTI_FILE_PIECE: { try { return handleTransferFilePiece( - PipeTransferTsFilePieceWithModReq.fromTPipeTransferReq(req), + PipeTransferMultiFilePieceReq.fromTPipeTransferReq(req), req instanceof AirGapPseudoTPipeTransferRequest, false); @@ -364,7 +364,7 @@ { try { return handleTransferFileSealV2( - PipeTransferTsFileSealWithModReq.fromTPipeTransferReq(req)); + PipeTransferMultiFileSealReq.fromTPipeTransferReq(req)); } finally { PipeDataNodeReceiverMetrics.getInstance() .recordTransferTsFileSealWithModTimer(System.nanoTime() - startTime); @@ -557,14 +557,14 @@ protected TSStatus loadFileV2( final PipeTransferFileSealReqV2 req, final List<String> fileAbsolutePaths) throws IOException, IllegalPathException { - return req instanceof PipeTransferTsFileSealWithModReq + return req instanceof PipeTransferMultiFileSealReq // TsFile's absolute path will be the second element ? (isUsingAsyncLoadTsFileStrategy.get() ? loadTsFileAsync( - ((PipeTransferTsFileSealWithModReq) req).getDatabaseNameByTsFileName(), + ((PipeTransferMultiFileSealReq) req).getDatabaseNameByTsFileName(), fileAbsolutePaths) : loadTsFileSync( - ((PipeTransferTsFileSealWithModReq) req).getDatabaseNameByTsFileName(), + ((PipeTransferMultiFileSealReq) req).getDatabaseNameByTsFileName(), fileAbsolutePaths.get(req.getFileNames().size() - 1))) : loadSchemaSnapShot(req.getParameters(), fileAbsolutePaths); }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferMultiFilePieceReq.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferMultiFilePieceReq.java new file mode 100644 index 0000000..96637f7 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferMultiFilePieceReq.java
@@ -0,0 +1,91 @@ +/* + * 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.pipe.sink.payload.evolvable.request; + +import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType; +import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferFilePieceReq; +import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; + +import java.io.IOException; +import java.nio.ByteBuffer; + +public class PipeTransferMultiFilePieceReq extends PipeTransferFilePieceReq { + + private PipeTransferMultiFilePieceReq() { + // Empty constructor + } + + @Override + protected PipeRequestType getPlanType() { + return PipeRequestType.TRANSFER_MULTI_FILE_PIECE; + } + + /////////////////////////////// Thrift /////////////////////////////// + + public static PipeTransferMultiFilePieceReq toTPipeTransferReq( + final String fileName, final long startWritingOffset, final byte[] filePiece) + throws IOException { + return (PipeTransferMultiFilePieceReq) + new PipeTransferMultiFilePieceReq() + .convertToTPipeTransferReq(fileName, startWritingOffset, filePiece); + } + + public static PipeTransferMultiFilePieceReq toTPipeTransferReq( + final String fileName, final ByteBuffer filePiece, final int transferSize) + throws IOException { + return (PipeTransferMultiFilePieceReq) + new PipeTransferMultiFilePieceReq() + .convertToTPipeTransferReq(fileName, filePiece, transferSize); + } + + public static PipeTransferMultiFilePieceReq fromTPipeTransferReq( + final TPipeTransferReq transferReq) { + return (PipeTransferMultiFilePieceReq) + new PipeTransferMultiFilePieceReq().translateFromTPipeTransferReq(transferReq); + } + + /////////////////////////////// Air Gap /////////////////////////////// + + public static byte[] toTPipeTransferBytes( + final String fileName, final long startWritingOffset, final byte[] filePiece) + throws IOException { + return new PipeTransferMultiFilePieceReq() + .convertToTPipeTransferBytes(fileName, startWritingOffset, filePiece); + } + + public static byte[] toTPipeTransferBytes( + final String fileName, final ByteBuffer filePiece, final int transferSize) + throws IOException { + return new PipeTransferMultiFilePieceReq() + .convertToTPipeTransferBytes(fileName, filePiece, transferSize); + } + + /////////////////////////////// Object /////////////////////////////// + + @Override + public boolean equals(final Object obj) { + return obj instanceof PipeTransferMultiFilePieceReq && super.equals(obj); + } + + @Override + public int hashCode() { + return super.hashCode(); + } +}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferMultiFileSealReq.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferMultiFileSealReq.java new file mode 100644 index 0000000..f25fc6c --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferMultiFileSealReq.java
@@ -0,0 +1,144 @@ +/* + * 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.pipe.sink.payload.evolvable.request; + +import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType; +import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferFileSealReqV2; +import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +public class PipeTransferMultiFileSealReq extends PipeTransferFileSealReqV2 { + + private PipeTransferMultiFileSealReq() { + // Empty constructor + } + + @Override + protected PipeRequestType getPlanType() { + return PipeRequestType.TRANSFER_TS_FILE_SEAL_WITH_MOD; + } + + protected static final String DATABASE_NAME_KEY_PREFIX = "DATABASE_NAME_"; + + public String getDatabaseNameByTsFileName() { + return parameters == null + ? null + : parameters.get(generateDatabaseNameWithFileNameKey(fileNames.get(fileNames.size() - 1))); + } + + protected static String generateDatabaseNameWithFileNameKey(final String fileName) { + return DATABASE_NAME_KEY_PREFIX + fileName; + } + + /////////////////////////////// Thrift /////////////////////////////// + + public static PipeTransferMultiFileSealReq toTPipeTransferReq( + final String modFileName, + final long modFileLength, + final String tsFileName, + final long tsFileLength) + throws IOException { + // Sevo file depending on incoming adaptations + return toTPipeTransferReq(modFileName, modFileLength, null, 0L, tsFileName, tsFileLength, null); + } + + public static PipeTransferMultiFileSealReq toTPipeTransferReq( + final String modFileName, + final long modFileLength, + final String sevoFileName, + final long sevoFileLength, + final String tsFileName, + final long tsFileLength, + final String dataBaseName) + throws IOException { + final List<String> fileNames = new ArrayList<>(); + final List<Long> fileLengths = new ArrayList<>(); + if (Objects.nonNull(modFileName)) { + fileNames.add(modFileName); + fileLengths.add(modFileLength); + } + if (Objects.nonNull(sevoFileName)) { + fileNames.add(sevoFileName); + fileLengths.add(sevoFileLength); + } + fileNames.add(tsFileName); + fileLengths.add(tsFileLength); + return (PipeTransferMultiFileSealReq) + new PipeTransferMultiFileSealReq() + .convertToTPipeTransferReq( + fileNames, + fileLengths, + Collections.singletonMap( + generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName)); + } + + public static PipeTransferMultiFileSealReq fromTPipeTransferReq(final TPipeTransferReq req) { + return (PipeTransferMultiFileSealReq) + new PipeTransferMultiFileSealReq().translateFromTPipeTransferReq(req); + } + + /////////////////////////////// Air Gap /////////////////////////////// + + public static byte[] toTPipeTransferBytes( + final String modFileName, + final long modFileLength, + final String sevoFileName, + final long sevoFileLength, + final String tsFileName, + final long tsFileLength, + final String dataBaseName) + throws IOException { + final List<String> fileNames = new ArrayList<>(); + final List<Long> fileLengths = new ArrayList<>(); + if (Objects.nonNull(modFileName)) { + fileNames.add(modFileName); + fileLengths.add(modFileLength); + } + if (Objects.nonNull(sevoFileName)) { + fileNames.add(sevoFileName); + fileLengths.add(sevoFileLength); + } + fileNames.add(tsFileName); + fileLengths.add(tsFileLength); + return new PipeTransferMultiFileSealReq() + .convertToTPipeTransferSnapshotSealBytes( + fileNames, + fileLengths, + Collections.singletonMap( + generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName)); + } + + /////////////////////////////// Object /////////////////////////////// + + @Override + public boolean equals(final Object obj) { + return obj instanceof PipeTransferMultiFileSealReq && super.equals(obj); + } + + @Override + public int hashCode() { + return super.hashCode(); + } +}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFilePieceReq.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFilePieceReq.java index 8e62156..0435d9a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFilePieceReq.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFilePieceReq.java
@@ -24,7 +24,6 @@ import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; import java.io.IOException; -import java.nio.ByteBuffer; public class PipeTransferTsFilePieceReq extends PipeTransferFilePieceReq { @@ -46,13 +45,6 @@ .convertToTPipeTransferReq(fileName, startWritingOffset, filePiece); } - public static PipeTransferTsFilePieceReq toTPipeTransferReq( - String fileName, long startWritingOffset, ByteBuffer filePiece) throws IOException { - return (PipeTransferTsFilePieceReq) - new PipeTransferTsFilePieceReq() - .convertToTPipeTransferReq(fileName, startWritingOffset, filePiece); - } - public static PipeTransferTsFilePieceReq fromTPipeTransferReq(TPipeTransferReq transferReq) { return (PipeTransferTsFilePieceReq) new PipeTransferTsFilePieceReq().translateFromTPipeTransferReq(transferReq);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFilePieceWithModReq.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFilePieceWithModReq.java deleted file mode 100644 index 86255fa..0000000 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFilePieceWithModReq.java +++ /dev/null
@@ -1,81 +0,0 @@ -/* - * 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.pipe.sink.payload.evolvable.request; - -import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType; -import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferFilePieceReq; -import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; - -import java.io.IOException; -import java.nio.ByteBuffer; - -public class PipeTransferTsFilePieceWithModReq extends PipeTransferFilePieceReq { - - private PipeTransferTsFilePieceWithModReq() { - // Empty constructor - } - - @Override - protected PipeRequestType getPlanType() { - return PipeRequestType.TRANSFER_TS_FILE_PIECE_WITH_MOD; - } - - /////////////////////////////// Thrift /////////////////////////////// - - public static PipeTransferTsFilePieceWithModReq toTPipeTransferReq( - String fileName, long startWritingOffset, byte[] filePiece) throws IOException { - return (PipeTransferTsFilePieceWithModReq) - new PipeTransferTsFilePieceWithModReq() - .convertToTPipeTransferReq(fileName, startWritingOffset, filePiece); - } - - public static PipeTransferTsFilePieceWithModReq toTPipeTransferReq( - String fileName, long startWritingOffset, ByteBuffer filePiece) throws IOException { - return (PipeTransferTsFilePieceWithModReq) - new PipeTransferTsFilePieceWithModReq() - .convertToTPipeTransferReq(fileName, startWritingOffset, filePiece); - } - - public static PipeTransferTsFilePieceWithModReq fromTPipeTransferReq( - TPipeTransferReq transferReq) { - return (PipeTransferTsFilePieceWithModReq) - new PipeTransferTsFilePieceWithModReq().translateFromTPipeTransferReq(transferReq); - } - - /////////////////////////////// Air Gap /////////////////////////////// - - public static byte[] toTPipeTransferBytes( - String fileName, long startWritingOffset, byte[] filePiece) throws IOException { - return new PipeTransferTsFilePieceWithModReq() - .convertToTPipeTransferBytes(fileName, startWritingOffset, filePiece); - } - - /////////////////////////////// Object /////////////////////////////// - - @Override - public boolean equals(Object obj) { - return obj instanceof PipeTransferTsFilePieceWithModReq && super.equals(obj); - } - - @Override - public int hashCode() { - return super.hashCode(); - } -}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFileSealWithModReq.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFileSealWithModReq.java deleted file mode 100644 index 25fb874..0000000 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFileSealWithModReq.java +++ /dev/null
@@ -1,150 +0,0 @@ -/* - * 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.pipe.sink.payload.evolvable.request; - -import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType; -import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferFileSealReqV2; -import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; - -import java.io.IOException; -import java.util.Arrays; -import java.util.Collections; -import java.util.HashMap; - -public class PipeTransferTsFileSealWithModReq extends PipeTransferFileSealReqV2 { - - private PipeTransferTsFileSealWithModReq() { - // Empty constructor - } - - @Override - protected PipeRequestType getPlanType() { - return PipeRequestType.TRANSFER_TS_FILE_SEAL_WITH_MOD; - } - - protected static final String DATABASE_NAME_KEY_PREFIX = "DATABASE_NAME_"; - - public String getDatabaseNameByTsFileName() { - return parameters == null - ? null - : parameters.get(generateDatabaseNameWithFileNameKey(fileNames.get(fileNames.size() - 1))); - } - - protected static String generateDatabaseNameWithFileNameKey(final String fileName) { - return DATABASE_NAME_KEY_PREFIX + fileName; - } - - /////////////////////////////// Thrift /////////////////////////////// - - public static PipeTransferTsFileSealWithModReq toTPipeTransferReq( - final String modFileName, - final long modFileLength, - final String tsFileName, - final long tsFileLength) - throws IOException { - return toTPipeTransferReq(modFileName, modFileLength, tsFileName, tsFileLength, null); - } - - public static PipeTransferTsFileSealWithModReq toTPipeTransferReq( - final String modFileName, - final long modFileLength, - final String tsFileName, - final long tsFileLength, - final String dataBaseName) - throws IOException { - return (PipeTransferTsFileSealWithModReq) - new PipeTransferTsFileSealWithModReq() - .convertToTPipeTransferReq( - Arrays.asList(modFileName, tsFileName), - Arrays.asList(modFileLength, tsFileLength), - Collections.singletonMap( - generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName)); - } - - public static PipeTransferTsFileSealWithModReq toTPipeTransferReq( - final String tsFileName, final long tsFileLength, final String dataBaseName) - throws IOException { - return (PipeTransferTsFileSealWithModReq) - new PipeTransferTsFileSealWithModReq() - .convertToTPipeTransferReq( - Collections.singletonList(tsFileName), - Collections.singletonList(tsFileLength), - Collections.singletonMap( - generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName)); - } - - public static PipeTransferTsFileSealWithModReq fromTPipeTransferReq(final TPipeTransferReq req) { - return (PipeTransferTsFileSealWithModReq) - new PipeTransferTsFileSealWithModReq().translateFromTPipeTransferReq(req); - } - - /////////////////////////////// Air Gap /////////////////////////////// - - public static byte[] toTPipeTransferBytes( - final String modFileName, - final long modFileLength, - final String tsFileName, - final long tsFileLength) - throws IOException { - return new PipeTransferTsFileSealWithModReq() - .convertToTPipeTransferSnapshotSealBytes( - Arrays.asList(modFileName, tsFileName), - Arrays.asList(modFileLength, tsFileLength), - new HashMap<>()); - } - - public static byte[] toTPipeTransferBytes( - final String modFileName, - final long modFileLength, - final String tsFileName, - final long tsFileLength, - final String dataBaseName) - throws IOException { - return new PipeTransferTsFileSealWithModReq() - .convertToTPipeTransferSnapshotSealBytes( - Arrays.asList(modFileName, tsFileName), - Arrays.asList(modFileLength, tsFileLength), - Collections.singletonMap( - generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName)); - } - - public static byte[] toTPipeTransferBytes( - final String tsFileName, final long tsFileLength, final String dataBaseName) - throws IOException { - return new PipeTransferTsFileSealWithModReq() - .convertToTPipeTransferSnapshotSealBytes( - Collections.singletonList(tsFileName), - Collections.singletonList(tsFileLength), - Collections.singletonMap( - generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName)); - } - - /////////////////////////////// Object /////////////////////////////// - - @Override - public boolean equals(final Object obj) { - return obj instanceof PipeTransferTsFileSealWithModReq && super.equals(obj); - } - - @Override - public int hashCode() { - return super.hashCode(); - } -}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java index 5bee20c..bdc4ca9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.db.pipe.sink.protocol.airgap; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.sink.limiter.TsFileSendRateLimiter; import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent; @@ -30,14 +31,17 @@ import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics; import org.apache.iotdb.db.pipe.metric.sink.PipeDataRegionSinkMetrics; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferMultiFilePieceReq; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferMultiFileSealReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferPlanNodeReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletBinaryReqV2; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletInsertNodeReqV2; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletRawReqV2; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceWithModReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealWithModReq; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.evolution.EvolvedSchema; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.evolution.SchemaEvolutionFile; import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException; import org.apache.iotdb.pipe.api.annotation.TableModel; import org.apache.iotdb.pipe.api.annotation.TreeModel; @@ -55,6 +59,7 @@ import java.io.File; import java.io.IOException; +import java.nio.ByteBuffer; import java.util.Arrays; import java.util.Objects; @@ -329,55 +334,79 @@ final String pipeName = pipeTsFileInsertionEvent.getPipeName(); final long creationTime = pipeTsFileInsertionEvent.getCreationTime(); final File tsFile = pipeTsFileInsertionEvent.getTsFile(); + final TsFileResource resource = pipeTsFileInsertionEvent.getResource(); final String errorMessage = String.format("Seal file %s error. Socket %s.", tsFile, socket); // 1. Transfer file piece by piece, and mod if needed - if (pipeTsFileInsertionEvent.isWithMod() && supportModsIfIsDataNodeReceiver) { - final File modFile = pipeTsFileInsertionEvent.getModFile(); - transferFilePieces(pipeName, creationTime, modFile, socket, true); + final boolean withMod = pipeTsFileInsertionEvent.isWithMod() && supportModsIfIsDataNodeReceiver; + final File modFile = pipeTsFileInsertionEvent.getModFile(); + transferFilePieces(pipeName, creationTime, modFile, socket, true); + if (withMod) { transferFilePieces(pipeName, creationTime, tsFile, socket, true); - // 2. Transfer file seal signal with mod, which means the file is transferred completely - if (!send( - pipeName, - creationTime, - socket, - PipeTransferTsFileSealWithModReq.toTPipeTransferBytes( - modFile.getName(), - modFile.length(), - tsFile.getName(), - tsFile.length(), - pipeTsFileInsertionEvent.isTableModelEvent() - ? pipeTsFileInsertionEvent.getTableModelDatabaseName() - : null))) { - receiverStatusHandler.handle( - new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode()) - .setMessage(errorMessage), - errorMessage, - pipeTsFileInsertionEvent.toString()); - } else { - LOGGER.info("Successfully transferred file {}.", tsFile); + } + + // 2. Transfer schema evolution file if exists + EvolvedSchema evolvedSchema = null; + ByteBuffer fileBuffer = null; + if (resource != null) { + evolvedSchema = resource.getMergedEvolvedSchema(); + if (evolvedSchema != null) { + fileBuffer = evolvedSchema.toSchemaEvolutionFileBuffer(); + final String sevoName = + SchemaEvolutionFile.getTsFileAssociatedSchemaEvolutionFileName(tsFile); + + while (fileBuffer.remaining() > 0) { + final int length = + Math.min( + PipeConfig.getInstance().getPipeSinkReadFileBufferSize(), fileBuffer.remaining()); + mayLimitRateAndRecordIO(length); + + if (!send( + pipeName, + creationTime, + socket, + PipeTransferMultiFilePieceReq.toTPipeTransferBytes(sevoName, fileBuffer, length))) { + final String sevoMessage = + String.format("Transfer file %s error. Socket %s.", sevoName, socket); + if (mayNeedHandshakeWhenFail()) { + // Send handshake because we don't know whether the receiver side configNode + // has set up a new one + sendHandshakeReq(socket); + } + receiverStatusHandler.handle( + new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode()) + .setMessage(sevoMessage), + sevoMessage, + tsFile.getName()); + } + } } + } + + // 3. Transfer file seal signal with mod, which means the file is transferred completely + if (!send( + pipeName, + creationTime, + socket, + PipeTransferMultiFileSealReq.toTPipeTransferBytes( + withMod ? modFile.getName() : null, + withMod ? modFile.length() : 0L, + Objects.nonNull(evolvedSchema) + ? SchemaEvolutionFile.getTsFileAssociatedSchemaEvolutionFileName(tsFile) + : null, + Objects.nonNull(evolvedSchema) ? fileBuffer.limit() : 0L, + tsFile.getName(), + tsFile.length(), + pipeTsFileInsertionEvent.isTableModelEvent() + ? pipeTsFileInsertionEvent.getTableModelDatabaseName() + : null))) { + receiverStatusHandler.handle( + new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode()) + .setMessage(errorMessage), + errorMessage, + pipeTsFileInsertionEvent.toString()); } else { - transferFilePieces(pipeName, creationTime, tsFile, socket, false); - // 2. Transfer file seal signal without mod, which means the file is transferred completely - if (!send( - pipeName, - creationTime, - socket, - PipeTransferTsFileSealWithModReq.toTPipeTransferBytes( - tsFile.getName(), - tsFile.length(), - pipeTsFileInsertionEvent.isTableModelEvent() - ? pipeTsFileInsertionEvent.getTableModelDatabaseName() - : null))) { - receiverStatusHandler.handle( - new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode()) - .setMessage(errorMessage), - errorMessage, - pipeTsFileInsertionEvent.toString()); - } else { - LOGGER.info("Successfully transferred file {}.", tsFile); - } + LOGGER.info("Successfully transferred file {}.", tsFile); } } @@ -398,7 +427,7 @@ @Override protected byte[] getTransferMultiFilePieceBytes( final String fileName, final long position, final byte[] payLoad) throws IOException { - return PipeTransferTsFilePieceWithModReq.toTPipeTransferBytes(fileName, position, payLoad); + return PipeTransferMultiFilePieceReq.toTPipeTransferBytes(fileName, position, payLoad); } @Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java index d95513a..52c52b1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java
@@ -69,7 +69,7 @@ public void transfer(final AsyncPipeDataTransferServiceClient client) throws TException { for (final Map.Entry<Pair<String, Long>, Long> entry : pipeName2BytesAccumulated.entrySet()) { - connector.rateLimitIfNeeded( + sink.rateLimitIfNeeded( entry.getKey().getLeft(), entry.getKey().getRight(), client.getEndPoint(), @@ -92,13 +92,11 @@ // Only handle the failed statuses to avoid string format performance overhead if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { - connector - .statusHandler() - .handle(status, response.getStatus().getMessage(), events.toString()); + sink.statusHandler().handle(status, response.getStatus().getMessage(), events.toString()); } for (final Pair<String, TEndPoint> redirectPair : LeaderCacheUtils.parseRecommendedRedirections(status)) { - connector.updateLeaderCache(redirectPair.getLeft(), redirectPair.getRight()); + sink.updateLeaderCache(redirectPair.getLeft(), redirectPair.getRight()); } events.forEach( @@ -123,7 +121,7 @@ events.size(), events.stream().map(EnrichedEvent::getPipeName).collect(Collectors.toSet())); } finally { - connector.addFailureEventsToRetryQueue(events, exception); + sink.addFailureEventsToRetryQueue(events, exception); } }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertNodeEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertNodeEventHandler.java index 70ba7f4..912a1e7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertNodeEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertNodeEventHandler.java
@@ -46,7 +46,7 @@ @Override protected void updateLeaderCache(final TSStatus status) { - connector.updateLeaderCache( + sink.updateLeaderCache( ((PipeInsertNodeTabletInsertionEvent) event).getDeviceId(), status.getRedirectNode()); } }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java index ca92af9..ac25281 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java
@@ -52,7 +52,7 @@ } public void transfer(final AsyncPipeDataTransferServiceClient client) throws TException { - connector.rateLimitIfNeeded( + sink.rateLimitIfNeeded( event.getPipeName(), event.getCreationTime(), client.getEndPoint(), req.getBody().length); tryTransfer(client, req); @@ -71,8 +71,7 @@ // Only handle the failed statuses to avoid string format performance overhead if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { - connector - .statusHandler() + sink.statusHandler() .handle(response.getStatus(), response.getStatus().getMessage(), event.toString()); } event.decreaseReferenceCount(PipeTransferTabletInsertionEventHandler.class.getName(), true); @@ -98,7 +97,7 @@ event.getCommitterKey(), event.getCommitId()); } finally { - connector.addFailureEventToRetryQueue(event, exception); + sink.addFailureEventToRetryQueue(event, exception); } }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletRawEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletRawEventHandler.java index ff1daa0..74910d6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletRawEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletRawEventHandler.java
@@ -45,7 +45,7 @@ @Override protected void updateLeaderCache(final TSStatus status) { - connector.updateLeaderCache( + sink.updateLeaderCache( ((PipeRawTabletInsertionEvent) event).getDeviceId(), status.getRedirectNode()); } }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java index 21f7c14..a8b4a3b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java
@@ -36,18 +36,18 @@ implements AsyncMethodCallback<TPipeTransferResp>, AutoCloseable { private static final Logger LOGGER = LoggerFactory.getLogger(PipeTransferTsFileHandler.class); - protected final IoTDBDataRegionAsyncSink connector; + protected final IoTDBDataRegionAsyncSink sink; protected volatile AsyncPipeDataTransferServiceClient client; - public PipeTransferTrackableHandler(final IoTDBDataRegionAsyncSink connector) { - this.connector = connector; + public PipeTransferTrackableHandler(final IoTDBDataRegionAsyncSink sink) { + this.sink = sink; } @Override public void onComplete(final TPipeTransferResp response) { - if (connector.isClosed()) { + if (sink.isClosed()) { clearEventsReferenceCount(); - connector.eliminateHandler(this, true); + sink.eliminateHandler(this, true); return; } @@ -56,7 +56,7 @@ // completed // NOTE: We should not clear the reference count of events, as this would cause the // `org.apache.iotdb.pipe.it.dual.tablemodel.manual.basic.IoTDBPipeDataSinkIT#testSinkTsFileFormat3` test to fail. - connector.eliminateHandler(this, false); + sink.eliminateHandler(this, false); } } @@ -67,14 +67,14 @@ client.setPrintLogWhenEncounterException(false); } - if (connector.isClosed()) { + if (sink.isClosed()) { clearEventsReferenceCount(); - connector.eliminateHandler(this, true); + sink.eliminateHandler(this, true); return; } onErrorInternal(exception); - connector.eliminateHandler(this, false); + sink.eliminateHandler(this, false); } /** @@ -93,10 +93,10 @@ this.client = client; } // track handler before checking if connector is closed - connector.trackHandler(this); - if (connector.isClosed()) { + sink.trackHandler(this); + if (sink.isClosed()) { clearEventsReferenceCount(); - connector.eliminateHandler(this, true); + sink.eliminateHandler(this, true); client.setShouldReturnSelf(true); client.returnSelf( (e) -> {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java index 2499b99..39a6a37 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
@@ -33,9 +33,9 @@ import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock; import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeAsyncClientManager; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferMultiFilePieceReq; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferMultiFileSealReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceWithModReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealWithModReq; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.evolution.EvolvedSchema; @@ -79,6 +79,7 @@ private final File tsFile; private final File modFile; private File currentFile; + private ByteBuffer sevoBuffer; private final TsFileResource resource; private final boolean transferMod; @@ -92,14 +93,14 @@ private RandomAccessFile reader; - private volatile boolean isSevoTransferStarted = false; + private volatile boolean needTransferSevo = false; private volatile boolean isSevoTransferred = false; private final AtomicBoolean isSealSignalSent; private IoTDBDataNodeAsyncClientManager clientManager; public PipeTransferTsFileHandler( - final IoTDBDataRegionAsyncSink connector, + final IoTDBDataRegionAsyncSink sink, final Map<Pair<String, Long>, Double> pipeName2WeightMap, final List<EnrichedEvent> events, final AtomicInteger eventsReferenceCount, @@ -110,7 +111,7 @@ final boolean transferMod, final String dataBaseName) throws InterruptedException { - super(connector); + super(sink); this.pipeName2WeightMap = pipeName2WeightMap; @@ -174,7 +175,7 @@ if (client == null) { LOGGER.warn( "Client has been returned to the pool. Current handler status is {}. Will not transfer {}.", - connector.isClosed() ? "CLOSED" : "NOT CLOSED", + sink.isClosed() ? "CLOSED" : "NOT CLOSED", tsFile); return; } @@ -182,47 +183,45 @@ client.setShouldReturnSelf(false); client.setTimeoutDynamically(clientManager.getConnectionTimeout()); - PipeResourceMetrics.getInstance().recordDiskIO(readFileBufferSize); - if (connector.isEnableSendTsFileLimit()) { - TsFileSendRateLimiter.getInstance().acquire(readFileBufferSize); - } final int readLength = reader.read(readBuffer); if (readLength == -1) { if (currentFile == modFile) { - currentFile = tsFile; - position = 0; try { reader.close(); } catch (final IOException e) { LOGGER.warn("Failed to close file reader when successfully transferred mod file.", e); } - reader = new RandomAccessFile(tsFile, "r"); - transfer(clientManager, client); - } else if (currentFile == tsFile) { + if (!isSevoTransferred && transferSevo(client)) { // if the transfer has been initiated, return directly to allow the callback to trigger // the next transfer return; } + currentFile = tsFile; + position = 0; + reader = new RandomAccessFile(tsFile, "r"); + transfer(clientManager, client); + } else if (currentFile == tsFile) { isSealSignalSent.set(true); final TPipeTransferReq uncompressedReq = - transferMod - ? PipeTransferTsFileSealWithModReq.toTPipeTransferReq( - modFile.getName(), - modFile.length(), - tsFile.getName(), - tsFile.length(), - dataBaseName) - : PipeTransferTsFileSealWithModReq.toTPipeTransferReq( - tsFile.getName(), tsFile.length(), dataBaseName); - final TPipeTransferReq req = connector.compressIfNeeded(uncompressedReq); + PipeTransferMultiFileSealReq.toTPipeTransferReq( + transferMod ? modFile.getName() : null, + transferMod ? modFile.length() : 0L, + Objects.nonNull(sevoBuffer) + ? SchemaEvolutionFile.getTsFileAssociatedSchemaEvolutionFileName(tsFile) + : null, + Objects.nonNull(sevoBuffer) ? sevoBuffer.limit() : 0L, + tsFile.getName(), + tsFile.length(), + dataBaseName); + final TPipeTransferReq req = sink.compressIfNeeded(uncompressedReq); pipeName2WeightMap.forEach( (pipePair, weight) -> - connector.rateLimitIfNeeded( + sink.rateLimitIfNeeded( pipePair.getLeft(), pipePair.getRight(), client.getEndPoint(), @@ -239,17 +238,23 @@ readLength == readFileBufferSize ? readBuffer : Arrays.copyOfRange(readBuffer, 0, readLength); + + PipeResourceMetrics.getInstance().recordDiskIO(payload.length); + if (sink.isEnableSendTsFileLimit()) { + TsFileSendRateLimiter.getInstance().acquire(payload.length); + } + final TPipeTransferReq uncompressedReq = - transferMod - ? PipeTransferTsFilePieceWithModReq.toTPipeTransferReq( + transferMod || needTransferSevo + ? PipeTransferMultiFilePieceReq.toTPipeTransferReq( currentFile.getName(), position, payload) : PipeTransferTsFilePieceReq.toTPipeTransferReq( currentFile.getName(), position, payload); - final TPipeTransferReq req = connector.compressIfNeeded(uncompressedReq); + final TPipeTransferReq req = sink.compressIfNeeded(uncompressedReq); pipeName2WeightMap.forEach( (pipePair, weight) -> - connector.rateLimitIfNeeded( + sink.rateLimitIfNeeded( pipePair.getLeft(), pipePair.getRight(), client.getEndPoint(), @@ -262,40 +267,45 @@ position += readLength; } + // Return iff the function has actually sent some data private boolean transferSevo(AsyncPipeDataTransferServiceClient client) throws IOException, TException { - isSevoTransferStarted = true; + if (Objects.isNull(sevoBuffer)) { + if (resource == null) { + isSevoTransferred = true; + // transferring tsfile written from tablets, no schema evolution + return false; + } - if (resource == null) { - isSevoTransferred = true; - // transferring tsfile written from tablets, no schema evolution - return false; + final EvolvedSchema evolvedSchema = resource.getMergedEvolvedSchema(); + if (evolvedSchema == null) { + isSevoTransferred = true; + return false; + } + + sevoBuffer = evolvedSchema.toSchemaEvolutionFileBuffer(); + needTransferSevo = true; + LOGGER.info("Transferring schema evolution file for tsfile {}.", tsFile); } - - EvolvedSchema evolvedSchema = resource.getMergedEvolvedSchema(); - if (evolvedSchema == null) { - isSevoTransferred = true; - return false; - } - - ByteBuffer fileBuffer = evolvedSchema.toSchemaEvolutionFileBuffer(); final TPipeTransferReq uncompressedReq = - PipeTransferTsFilePieceReq.toTPipeTransferReq( + PipeTransferMultiFilePieceReq.toTPipeTransferReq( SchemaEvolutionFile.getTsFileAssociatedSchemaEvolutionFileName(currentFile), - 0, - fileBuffer); - final TPipeTransferReq req = connector.compressIfNeeded(uncompressedReq); + sevoBuffer, + readFileBufferSize); + final TPipeTransferReq req = sink.compressIfNeeded(uncompressedReq); pipeName2WeightMap.forEach( (pipePair, weight) -> - connector.rateLimitIfNeeded( + sink.rateLimitIfNeeded( pipePair.getLeft(), pipePair.getRight(), client.getEndPoint(), (long) (req.getBody().length * weight))); + if (sink.isEnableSendTsFileLimit()) { + TsFileSendRateLimiter.getInstance().acquire(req.getBody().length); + } tryTransfer(client, req); - LOGGER.info("Transferring schema evolution file for tsfile {}.", tsFile); return true; } @@ -304,7 +314,7 @@ try { super.onComplete(response); } finally { - if (connector.isClosed()) { + if (sink.isClosed()) { returnClientIfNecessary(); } } @@ -318,8 +328,7 @@ // Only handle the failed statuses to avoid string format performance overhead if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { - connector - .statusHandler() + sink.statusHandler() .handle( status, String.format( @@ -393,16 +402,13 @@ // Only handle the failed statuses to avoid string format performance overhead if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { - connector - .statusHandler() - .handle(status, response.getStatus().getMessage(), tsFile.getName()); + sink.statusHandler().handle(status, response.getStatus().getMessage(), tsFile.getName()); } } - if (isSevoTransferStarted && !isSevoTransferred) { - isSevoTransferred = true; + if (Objects.nonNull(sevoBuffer)) { + isSevoTransferred = sevoBuffer.remaining() == 0; } - transfer(clientManager, client); } catch (final Exception e) { onError(e); @@ -471,7 +477,7 @@ returnClientIfNecessary(); } finally { if (eventsHadBeenAddedToRetryQueue.compareAndSet(false, true)) { - connector.addFailureEventsToRetryQueue(events, exception); + sink.addFailureEventsToRetryQueue(events, exception); } } } @@ -482,7 +488,7 @@ return; } - if (connector.isClosed()) { + if (sink.isClosed()) { closeClient(); } @@ -506,7 +512,7 @@ if (client == null) { LOGGER.warn( "Client has been returned to the pool. Current handler status is {}. Will not transfer {}.", - connector.isClosed() ? "CLOSED" : "NOT CLOSED", + sink.isClosed() ? "CLOSED" : "NOT CLOSED", tsFile); return; }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java index db8c210..4635b70 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
@@ -21,6 +21,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient; import org.apache.iotdb.commons.pipe.sink.limiter.TsFileSendRateLimiter; @@ -39,13 +40,13 @@ import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventPlainBatch; 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.pipe.sink.payload.evolvable.request.PipeTransferMultiFilePieceReq; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferMultiFileSealReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferPlanNodeReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletBinaryReqV2; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletInsertNodeReqV2; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletRawReqV2; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceWithModReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealWithModReq; import org.apache.iotdb.db.pipe.sink.util.cacher.LeaderCacheUtils; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -120,7 +121,7 @@ @Override protected PipeTransferFilePieceReq getTransferMultiFilePieceReq( final String fileName, final long position, final byte[] payLoad) throws IOException { - return PipeTransferTsFilePieceWithModReq.toTPipeTransferReq(fileName, position, payLoad); + return PipeTransferMultiFilePieceReq.toTPipeTransferReq(fileName, position, payLoad); } @Override @@ -530,22 +531,31 @@ TPipeTransferResp resp; // 1. Transfer tsFile, and mod file if exists and receiver's version >= 2 - if (Objects.nonNull(modFile) && clientManager.supportModsIfIsDataNodeReceiver()) { - transferFilePieces(pipeName2WeightMap, modFile, clientAndStatus, true); + final boolean withMod = + Objects.nonNull(modFile) && clientManager.supportModsIfIsDataNodeReceiver(); + transferFilePieces(pipeName2WeightMap, modFile, clientAndStatus, true); + if (withMod) { transferFilePieces(pipeName2WeightMap, tsFile, clientAndStatus, true); + } - try { - // 2. Transfer schema evolution file if exists - if (resource != null) { - EvolvedSchema evolvedSchema = resource.getMergedEvolvedSchema(); - if (evolvedSchema != null) { - ByteBuffer fileBuffer = evolvedSchema.toSchemaEvolutionFileBuffer(); - final byte[] payload = fileBuffer.array(); + try { + // 2. Transfer schema evolution file if exists + EvolvedSchema evolvedSchema = null; + ByteBuffer fileBuffer = null; + if (resource != null) { + evolvedSchema = resource.getMergedEvolvedSchema(); + if (evolvedSchema != null) { + fileBuffer = evolvedSchema.toSchemaEvolutionFileBuffer(); + final String sevoName = + SchemaEvolutionFile.getTsFileAssociatedSchemaEvolutionFileName(tsFile); + + while (fileBuffer.remaining() > 0) { + final int length = + Math.min( + PipeConfig.getInstance().getPipeSinkReadFileBufferSize(), + fileBuffer.remaining()); final TPipeTransferReq uncompressedReq = - PipeTransferTsFilePieceReq.toTPipeTransferReq( - SchemaEvolutionFile.getTsFileAssociatedSchemaEvolutionFileName(tsFile), - 0, - payload); + PipeTransferMultiFilePieceReq.toTPipeTransferReq(sevoName, fileBuffer, length); final TPipeTransferReq sevoReq = compressIfNeeded(uncompressedReq); pipeName2WeightMap.forEach( @@ -555,6 +565,7 @@ pipePair.getRight(), clientAndStatus.getLeft().getEndPoint(), (long) (sevoReq.getBody().length * weight))); + mayLimitRateAndRecordIO(length); resp = clientAndStatus.left.pipeTransfer(sevoReq); @@ -564,67 +575,44 @@ && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { receiverStatusHandler.handle( resp.getStatus(), - String.format("Seal file %s error, result status %s.", tsFile, resp.getStatus()), + String.format( + "Seal file %s error, result status %s.", sevoName, resp.getStatus()), tsFile.getName()); return; } - LOGGER.info("Transferred schema evolution file for tsfile {}.", tsFile); } } - - // 3. Transfer file seal signal with mod, which means the file is transferred completely - final TPipeTransferReq req = - compressIfNeeded( - PipeTransferTsFileSealWithModReq.toTPipeTransferReq( - modFile.getName(), - modFile.length(), - tsFile.getName(), - tsFile.length(), - dataBaseName)); - - pipeName2WeightMap.forEach( - (pipePair, weight) -> - rateLimitIfNeeded( - pipePair.getLeft(), - pipePair.getRight(), - clientAndStatus.getLeft().getEndPoint(), - (long) (req.getBody().length * weight))); - - resp = clientAndStatus.getLeft().pipeTransfer(req); - - } catch (final Exception e) { - clientAndStatus.setRight(false); - clientManager.adjustTimeoutIfNecessary(e); - throw new PipeConnectionException( - String.format("Network error when seal file %s, because %s.", tsFile, e.getMessage()), - e); } - } else { - transferFilePieces(pipeName2WeightMap, tsFile, clientAndStatus, false); - // 2. Transfer file seal signal without mod, which means the file is transferred completely - try { - final TPipeTransferReq req = - compressIfNeeded( - PipeTransferTsFileSealWithModReq.toTPipeTransferReq( - tsFile.getName(), tsFile.length(), dataBaseName)); + // 3. Transfer file seal signal with mod, which means the file is transferred completely + final TPipeTransferReq req = + compressIfNeeded( + PipeTransferMultiFileSealReq.toTPipeTransferReq( + withMod ? modFile.getName() : null, + withMod ? modFile.length() : 0L, + Objects.nonNull(evolvedSchema) + ? SchemaEvolutionFile.getTsFileAssociatedSchemaEvolutionFileName(tsFile) + : null, + Objects.nonNull(evolvedSchema) ? fileBuffer.limit() : 0L, + tsFile.getName(), + tsFile.length(), + dataBaseName)); - pipeName2WeightMap.forEach( - (pipePair, weight) -> - rateLimitIfNeeded( - pipePair.getLeft(), - pipePair.getRight(), - clientAndStatus.getLeft().getEndPoint(), - (long) (req.getBody().length * weight))); + pipeName2WeightMap.forEach( + (pipePair, weight) -> + rateLimitIfNeeded( + pipePair.getLeft(), + pipePair.getRight(), + clientAndStatus.getLeft().getEndPoint(), + (long) (req.getBody().length * weight))); - resp = clientAndStatus.getLeft().pipeTransfer(req); - } catch (final Exception e) { - clientAndStatus.setRight(false); - clientManager.adjustTimeoutIfNecessary(e); - throw new PipeConnectionException( - String.format("Network error when seal file %s, because %s.", tsFile, e.getMessage()), - e); - } + resp = clientAndStatus.getLeft().pipeTransfer(req); + + } catch (final Exception e) { + clientAndStatus.setRight(false); + clientManager.adjustTimeoutIfNecessary(e); + throw new PipeConnectionException( + String.format("Network error when seal file %s, because %s.", tsFile, e.getMessage()), e); } final TSStatus status = resp.getStatus();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java index 0cc4470..0331bdf 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.pipe.sink.payload.thrift.response.PipeTransferFilePieceResp; import org.apache.iotdb.commons.schema.SchemaConstant; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferMultiFilePieceReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferPlanNodeReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferSchemaSnapshotPieceReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferSchemaSnapshotSealReq; @@ -35,7 +36,6 @@ import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletRawReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletRawReqV2; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceReq; -import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceWithModReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealReq; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.CreateAlignedTimeSeriesNode; @@ -489,10 +489,10 @@ final byte[] body = "testPipeTransferFilePieceWithModReq".getBytes(); final String fileName = "1.tsfile.mod"; - final PipeTransferTsFilePieceWithModReq req = - PipeTransferTsFilePieceWithModReq.toTPipeTransferReq(fileName, 0, body); - final PipeTransferTsFilePieceWithModReq deserializeReq = - PipeTransferTsFilePieceWithModReq.fromTPipeTransferReq(req); + final PipeTransferMultiFilePieceReq req = + PipeTransferMultiFilePieceReq.toTPipeTransferReq(fileName, 0, body); + final PipeTransferMultiFilePieceReq deserializeReq = + PipeTransferMultiFilePieceReq.fromTPipeTransferReq(req); Assert.assertEquals(req.getVersion(), deserializeReq.getVersion()); Assert.assertEquals(req.getType(), deserializeReq.getType());
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java index 97745bb..9b1b327 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
@@ -23,7 +23,6 @@ import org.apache.iotdb.commons.audit.IAuditEntity; import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.conf.CommonDescriptor; -import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; @@ -407,16 +406,41 @@ final PipeTransferFilePieceReq req, final boolean isRequestThroughAirGap, final boolean isSingleFile) { - String fileName = req.getFileName(); - String suffix = fileName.substring(fileName.lastIndexOf('.')); - try { - if (suffix.equals(IoTDBConstant.SCHEMA_EVOLUTION_FILE_SUFFIX)) { - handleTransferSevoFile(req); - return PipeTransferFilePieceResp.toTPipeTransferResp( - RpcUtils.SUCCESS_STATUS, writingFileWriter.length()); + updateWritingFileIfNeeded(req.getFileName(), isSingleFile); + + // If the request is through air gap, the sender will resend the file piece from the beginning + // of the file. So the receiver should reset the offset of the writing file to the beginning + // of the file. + if (isRequestThroughAirGap && req.getStartWritingOffset() < writingFileWriter.length()) { + writingFileWriter.setLength(req.getStartWritingOffset()); } - return handleTsFileOrMods(req, isRequestThroughAirGap, isSingleFile); + + if (!isWritingFileOffsetCorrect(req.getStartWritingOffset())) { + if (!writingFile.getName().endsWith(TsFileConstant.TSFILE_SUFFIX)) { + // If the file is a tsFile, then the content will not be changed for a specific filename. + // However, for other files (mod, snapshot, etc.) the content varies for the same name in + // different times, then we must rewrite the file to apply the newest version. + writingFileWriter.setLength(0); + } + + final TSStatus status = + RpcUtils.getStatus( + TSStatusCode.PIPE_TRANSFER_FILE_OFFSET_RESET, + String.format( + "Request sender to reset file reader's offset from %s to %s.", + req.getStartWritingOffset(), writingFileWriter.length())); + PipeLogger.log( + LOGGER::warn, + "Receiver id = %s: File offset reset requested by receiver, response status = %s.", + receiverId.get(), + status); + return PipeTransferFilePieceResp.toTPipeTransferResp(status, writingFileWriter.length()); + } + + writingFileWriter.write(req.getFilePiece()); + return PipeTransferFilePieceResp.toTPipeTransferResp( + RpcUtils.SUCCESS_STATUS, writingFileWriter.length()); } catch (final Exception e) { PipeLogger.log( LOGGER::warn, @@ -437,58 +461,6 @@ } } - private void handleTransferSevoFile(PipeTransferFilePieceReq req) throws IOException { - File file = new File(receiverFileDirWithIdSuffix.get(), req.getFileName()); - try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw")) { - randomAccessFile.write(req.getFilePiece()); - } - LOGGER.info( - "Receiver id = {}: written schema evolution file {} .", - receiverId.get(), - req.getFileName()); - } - - private PipeTransferFilePieceResp handleTsFileOrMods( - final PipeTransferFilePieceReq req, - final boolean isRequestThroughAirGap, - final boolean isSingleFile) - throws IOException { - updateWritingFileIfNeeded(req.getFileName(), isSingleFile); - - // If the request is through air gap, the sender will resend the file piece from the beginning - // of the file. So the receiver should reset the offset of the writing file to the beginning - // of the file. - if (isRequestThroughAirGap && req.getStartWritingOffset() < writingFileWriter.length()) { - writingFileWriter.setLength(req.getStartWritingOffset()); - } - - if (!isWritingFileOffsetCorrect(req.getStartWritingOffset())) { - if (!writingFile.getName().endsWith(TsFileConstant.TSFILE_SUFFIX)) { - // If the file is a tsFile, then the content will not be changed for a specific filename. - // However, for other files (mod, snapshot, etc.) the content varies for the same name in - // different times, then we must rewrite the file to apply the newest version. - writingFileWriter.setLength(0); - } - - final TSStatus status = - RpcUtils.getStatus( - TSStatusCode.PIPE_TRANSFER_FILE_OFFSET_RESET, - String.format( - "Request sender to reset file reader's offset from %s to %s.", - req.getStartWritingOffset(), writingFileWriter.length())); - PipeLogger.log( - LOGGER::warn, - "Receiver id = %s: File offset reset requested by receiver, response status = %s.", - receiverId.get(), - status); - return PipeTransferFilePieceResp.toTPipeTransferResp(status, writingFileWriter.length()); - } - - writingFileWriter.write(req.getFilePiece()); - return PipeTransferFilePieceResp.toTPipeTransferResp( - RpcUtils.SUCCESS_STATUS, writingFileWriter.length()); - } - protected final void updateWritingFileIfNeeded(final String fileName, final boolean isSingleFile) throws IOException { if (isFileExistedAndNameCorrect(fileName)) {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/thrift/request/PipeRequestType.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/thrift/request/PipeRequestType.java index 97a9526..ebee8ea 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/thrift/request/PipeRequestType.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/thrift/request/PipeRequestType.java
@@ -38,7 +38,7 @@ TRANSFER_TS_FILE_SEAL((short) 5), TRANSFER_TABLET_BATCH((short) 6), TRANSFER_TABLET_BINARY((short) 7), - TRANSFER_TS_FILE_PIECE_WITH_MOD((short) 8), + TRANSFER_MULTI_FILE_PIECE((short) 8), TRANSFER_TS_FILE_SEAL_WITH_MOD((short) 9), TRANSFER_TABLET_INSERT_NODE_V2((short) 10),
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/thrift/request/PipeTransferFilePieceReq.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/thrift/request/PipeTransferFilePieceReq.java index 7b87d0a..0096a58 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/thrift/request/PipeTransferFilePieceReq.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/thrift/request/PipeTransferFilePieceReq.java
@@ -74,22 +74,24 @@ } protected final PipeTransferFilePieceReq convertToTPipeTransferReq( - String snapshotName, long startWritingOffset, ByteBuffer snapshotPiece) throws IOException { + final String snapshotName, final ByteBuffer snapshotPiece, final int transferSize) + throws IOException { this.fileName = snapshotName; - this.startWritingOffset = startWritingOffset; + this.startWritingOffset = snapshotPiece.position(); this.filePiece = snapshotPiece.array(); this.version = IoTDBSinkRequestVersion.VERSION_1.getVersion(); this.type = getPlanType().getType(); try (final PublicBAOS byteArrayOutputStream = new PublicBAOS(); final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) { + final int size = Math.min(transferSize, snapshotPiece.remaining()); ReadWriteIOUtils.write(snapshotName, outputStream); ReadWriteIOUtils.write(startWritingOffset, outputStream); - ReadWriteIOUtils.write(snapshotPiece.remaining(), outputStream); + ReadWriteIOUtils.write(size, outputStream); ReadWriteIOUtils.writeWithoutSize( - snapshotPiece, snapshotPiece.position(), snapshotPiece.remaining(), outputStream); - snapshotPiece.position(snapshotPiece.position() + snapshotPiece.remaining()); + snapshotPiece, snapshotPiece.position(), size, outputStream); + snapshotPiece.position(snapshotPiece.position() + size); body = ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()); } @@ -124,6 +126,24 @@ } } + protected final byte[] convertToTPipeTransferBytes( + final String snapshotName, final ByteBuffer snapshotPiece, final int transferSize) + throws IOException { + try (final PublicBAOS byteArrayOutputStream = new PublicBAOS(); + final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) { + ReadWriteIOUtils.write(IoTDBSinkRequestVersion.VERSION_1.getVersion(), outputStream); + ReadWriteIOUtils.write(getPlanType().getType(), outputStream); + final int size = Math.min(transferSize, snapshotPiece.remaining()); + ReadWriteIOUtils.write(snapshotName, outputStream); + ReadWriteIOUtils.write(snapshotPiece.position(), outputStream); + ReadWriteIOUtils.write(size, outputStream); + ReadWriteIOUtils.writeWithoutSize( + snapshotPiece, snapshotPiece.position(), size, outputStream); + snapshotPiece.position(snapshotPiece.position() + size); + return byteArrayOutputStream.toByteArray(); + } + } + /////////////////////////////// Object /////////////////////////////// @Override
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java index 7d84e3b..0e453b4 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java
@@ -274,7 +274,6 @@ long position = 0; try (final RandomAccessFile reader = new RandomAccessFile(file, "r")) { while (true) { - mayLimitRateAndRecordIO(readFileBufferSize); final int readLength = reader.read(readBuffer); if (readLength == -1) { break; @@ -284,6 +283,8 @@ readLength == readFileBufferSize ? readBuffer : Arrays.copyOfRange(readBuffer, 0, readLength); + + mayLimitRateAndRecordIO(payload.length); if (!send( pipeName, creationTime,
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java index 75a4607..c634756 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java
@@ -189,7 +189,6 @@ long position = 0; try (final RandomAccessFile reader = new RandomAccessFile(file, "r")) { while (true) { - mayLimitRateAndRecordIO(readFileBufferSize); final int readLength = reader.read(readBuffer); if (readLength == -1) { break; @@ -199,6 +198,8 @@ readLength == readFileBufferSize ? readBuffer : Arrays.copyOfRange(readBuffer, 0, readLength); + mayLimitRateAndRecordIO(payLoad.length); + final PipeTransferFilePieceResp resp; try { final TPipeTransferReq req =