Skip to content

Commit 5282774

Browse files
CaideyipiJackieTien97
Caideyipi
authored and
JackieTien97
committed
Support legacy pipe receiver requests (#17901)
* Support legacy pipe snapshot seal parameters * Support legacy pipe tablet requests * Resolve legacy pipe tablet request conflicts * Add pipe receiver 1.3 compatibility tests * Try current raw tablet format first * Fix legacy raw tablet fallback
1 parent 6bcec43 commit 5282774

11 files changed

Lines changed: 1101 additions & 20 deletions

File tree

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1277,15 +1277,15 @@ protected TSStatus loadFileV2(
12771277
PipeConfigRegionSnapshotEvent.getConfigPhysicalPlanTypeSet(
12781278
parameters.get(ColumnHeaderConstant.TYPE));
12791279
final boolean isTreeModelDataAllowedToBeCaptured =
1280-
parameters.containsKey(PipeTransferFileSealReqV2.TREE);
1280+
PipeTransferFileSealReqV2.isTreeModelDataAllowedToBeCaptured(parameters);
12811281
final TreePattern treePattern =
12821282
TreePattern.parsePatternFromString(
12831283
parameters.get(ColumnHeaderConstant.PATH_PATTERN),
12841284
isTreeModelDataAllowedToBeCaptured,
12851285
p -> new IoTDBTreePattern(isTreeModelDataAllowedToBeCaptured, p));
12861286
final TablePattern tablePattern =
12871287
new TablePattern(
1288-
parameters.containsKey(PipeTransferFileSealReqV2.TABLE),
1288+
PipeTransferFileSealReqV2.isTableModelDataAllowedToBeCaptured(parameters),
12891289
parameters.get(PipeTransferFileSealReqV2.DATABASE_PATTERN),
12901290
parameters.get(ColumnHeaderConstant.TABLE_NAME));
12911291
final List<TSStatus> results = new ArrayList<>();

‎iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/sink/PipeConfigNodeThriftRequestTest.java‎

Lines changed: 212 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,17 +19,33 @@
1919

2020
package org.apache.iotdb.confignode.manager.pipe.sink;
2121

22+
import org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferHandshakeConstant;
23+
import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.IoTDBSinkRequestVersion;
24+
import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType;
25+
import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferFileSealReqV2;
26+
import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
2227
import org.apache.iotdb.confignode.consensus.request.write.cq.ActiveCQPlan;
2328
import org.apache.iotdb.confignode.manager.pipe.sink.payload.PipeTransferConfigNodeHandshakeV1Req;
29+
import org.apache.iotdb.confignode.manager.pipe.sink.payload.PipeTransferConfigNodeHandshakeV2Req;
2430
import org.apache.iotdb.confignode.manager.pipe.sink.payload.PipeTransferConfigPlanReq;
2531
import org.apache.iotdb.confignode.manager.pipe.sink.payload.PipeTransferConfigSnapshotPieceReq;
2632
import org.apache.iotdb.confignode.manager.pipe.sink.payload.PipeTransferConfigSnapshotSealReq;
2733
import org.apache.iotdb.confignode.persistence.schema.CNSnapshotFileType;
34+
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
2835

36+
import org.apache.tsfile.utils.Binary;
37+
import org.apache.tsfile.utils.PublicBAOS;
38+
import org.apache.tsfile.utils.ReadWriteIOUtils;
2939
import org.junit.Assert;
3040
import org.junit.Test;
3141

42+
import java.io.DataOutputStream;
3243
import java.io.IOException;
44+
import java.nio.ByteBuffer;
45+
import java.util.Arrays;
46+
import java.util.HashMap;
47+
import java.util.List;
48+
import java.util.Map;
3349

3450
public class PipeConfigNodeThriftRequestTest {
3551

@@ -48,6 +64,59 @@ public void testPipeTransferConfigHandshakeReq() throws IOException {
4864
Assert.assertEquals(req.getTimestampPrecision(), deserializeReq.getTimestampPrecision());
4965
}
5066

67+
@Test
68+
public void testPipeTransferConfigHandshakeReqFromLegacyV13Body() throws IOException {
69+
final TPipeTransferReq req =
70+
legacyTransferReq(
71+
PipeRequestType.HANDSHAKE_CONFIGNODE_V1,
72+
serializeLegacyHandshakeV1Body(TIME_PRECISION));
73+
74+
final PipeTransferConfigNodeHandshakeV1Req deserializeReq =
75+
PipeTransferConfigNodeHandshakeV1Req.fromTPipeTransferReq(req);
76+
77+
Assert.assertEquals(req.getVersion(), deserializeReq.getVersion());
78+
Assert.assertEquals(req.getType(), deserializeReq.getType());
79+
Assert.assertEquals(TIME_PRECISION, deserializeReq.getTimestampPrecision());
80+
}
81+
82+
@Test
83+
public void testPipeTransferConfigHandshakeV2Req() throws IOException {
84+
final Map<String, String> params = new HashMap<>();
85+
params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_TIME_PRECISION, TIME_PRECISION);
86+
params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_CLUSTER_ID, "cluster");
87+
params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_USERNAME, "root");
88+
params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_PASSWORD, "root");
89+
90+
final PipeTransferConfigNodeHandshakeV2Req req =
91+
PipeTransferConfigNodeHandshakeV2Req.toTPipeTransferReq(params);
92+
final PipeTransferConfigNodeHandshakeV2Req deserializeReq =
93+
PipeTransferConfigNodeHandshakeV2Req.fromTPipeTransferReq(req);
94+
95+
Assert.assertEquals(req.getVersion(), deserializeReq.getVersion());
96+
Assert.assertEquals(req.getType(), deserializeReq.getType());
97+
Assert.assertEquals(params, deserializeReq.getParams());
98+
}
99+
100+
@Test
101+
public void testPipeTransferConfigHandshakeV2ReqFromLegacyV13Body() throws IOException {
102+
final Map<String, String> params = new HashMap<>();
103+
params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_TIME_PRECISION, TIME_PRECISION);
104+
params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_CLUSTER_ID, "cluster");
105+
params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_USERNAME, "root");
106+
params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_PASSWORD, "root");
107+
108+
final TPipeTransferReq req =
109+
legacyTransferReq(
110+
PipeRequestType.HANDSHAKE_CONFIGNODE_V2, serializeLegacyHandshakeV2Body(params));
111+
112+
final PipeTransferConfigNodeHandshakeV2Req deserializeReq =
113+
PipeTransferConfigNodeHandshakeV2Req.fromTPipeTransferReq(req);
114+
115+
Assert.assertEquals(req.getVersion(), deserializeReq.getVersion());
116+
Assert.assertEquals(req.getType(), deserializeReq.getType());
117+
Assert.assertEquals(params, deserializeReq.getParams());
118+
}
119+
51120
@Test
52121
public void testPipeTransferConfigPlanReq() {
53122
PipeTransferConfigPlanReq req =
@@ -56,6 +125,22 @@ public void testPipeTransferConfigPlanReq() {
56125

57126
Assert.assertEquals(req.getVersion(), deserializeReq.getVersion());
58127
Assert.assertEquals(req.getType(), deserializeReq.getType());
128+
Assert.assertArrayEquals(req.getBody(), deserializeReq.getBody());
129+
}
130+
131+
@Test
132+
public void testPipeTransferConfigPlanReqFromLegacyV13Body() {
133+
final ActiveCQPlan plan = new ActiveCQPlan("cqId", "md5");
134+
final ByteBuffer legacyBody = plan.serializeToByteBuffer();
135+
final TPipeTransferReq req =
136+
legacyTransferReq(PipeRequestType.TRANSFER_CONFIG_PLAN, legacyBody);
137+
138+
final PipeTransferConfigPlanReq deserializeReq =
139+
PipeTransferConfigPlanReq.fromTPipeTransferReq(req);
140+
141+
Assert.assertEquals(req.getVersion(), deserializeReq.getVersion());
142+
Assert.assertEquals(req.getType(), deserializeReq.getType());
143+
Assert.assertArrayEquals(byteBufferToByteArray(legacyBody), deserializeReq.getBody());
59144
}
60145

61146
@Test
@@ -76,6 +161,24 @@ public void testPipeTransferConfigSnapshotPieceReq() throws IOException {
76161
Assert.assertArrayEquals(req.getFilePiece(), deserializeReq.getFilePiece());
77162
}
78163

164+
@Test
165+
public void testPipeTransferConfigSnapshotPieceReqFromLegacyV13Body() throws IOException {
166+
final byte[] body = "legacyConfigSnapshotPiece".getBytes();
167+
final TPipeTransferReq req =
168+
legacyTransferReq(
169+
PipeRequestType.TRANSFER_CONFIG_SNAPSHOT_PIECE,
170+
serializeLegacyFilePieceBody("cluster_schema.bin", 10L, body));
171+
172+
final PipeTransferConfigSnapshotPieceReq deserializeReq =
173+
PipeTransferConfigSnapshotPieceReq.fromTPipeTransferReq(req);
174+
175+
Assert.assertEquals(req.getVersion(), deserializeReq.getVersion());
176+
Assert.assertEquals(req.getType(), deserializeReq.getType());
177+
Assert.assertEquals("cluster_schema.bin", deserializeReq.getFileName());
178+
Assert.assertEquals(10L, deserializeReq.getStartWritingOffset());
179+
Assert.assertArrayEquals(body, deserializeReq.getFilePiece());
180+
}
181+
79182
@Test
80183
public void testPipeTransferConfigSnapshotSealReq() throws IOException {
81184
String snapshotName = "cluster_schema.bin";
@@ -108,4 +211,113 @@ public void testPipeTransferConfigSnapshotSealReq() throws IOException {
108211
Assert.assertEquals(req.getFileLengths(), deserializeReq.getFileLengths());
109212
Assert.assertEquals(req.getParameters(), deserializeReq.getParameters());
110213
}
214+
215+
@Test
216+
public void testPipeTransferConfigSnapshotSealReqFromLegacyV13Body() throws IOException {
217+
final String snapshotName = "cluster_schema.bin";
218+
final String templateInfoName = "template_info.bin";
219+
final Map<String, String> parameters = new HashMap<>();
220+
parameters.put(ColumnHeaderConstant.PATH_PATTERN, "root.**");
221+
parameters.put(
222+
PipeTransferConfigSnapshotSealReq.FILE_TYPE,
223+
Byte.toString(CNSnapshotFileType.SCHEMA.getType()));
224+
parameters.put(ColumnHeaderConstant.TYPE, "200");
225+
226+
final TPipeTransferReq req =
227+
legacyTransferReq(
228+
PipeRequestType.TRANSFER_CONFIG_SNAPSHOT_SEAL,
229+
serializeLegacyFileSealV2Body(
230+
Arrays.asList(snapshotName, templateInfoName),
231+
Arrays.asList(100L, 10L),
232+
parameters));
233+
final PipeTransferConfigSnapshotSealReq deserializeReq =
234+
PipeTransferConfigSnapshotSealReq.fromTPipeTransferReq(req);
235+
236+
Assert.assertEquals(req.getVersion(), deserializeReq.getVersion());
237+
Assert.assertEquals(req.getType(), deserializeReq.getType());
238+
Assert.assertEquals(
239+
Arrays.asList(snapshotName, templateInfoName), deserializeReq.getFileNames());
240+
Assert.assertEquals(Arrays.asList(100L, 10L), deserializeReq.getFileLengths());
241+
Assert.assertEquals(parameters, deserializeReq.getParameters());
242+
Assert.assertTrue(
243+
PipeTransferFileSealReqV2.isTreeModelDataAllowedToBeCaptured(
244+
deserializeReq.getParameters()));
245+
Assert.assertFalse(
246+
PipeTransferFileSealReqV2.isTableModelDataAllowedToBeCaptured(
247+
deserializeReq.getParameters()));
248+
}
249+
250+
private static TPipeTransferReq legacyTransferReq(
251+
final PipeRequestType requestType, final ByteBuffer body) {
252+
final TPipeTransferReq req = new TPipeTransferReq();
253+
req.version = IoTDBSinkRequestVersion.VERSION_1.getVersion();
254+
req.type = requestType.getType();
255+
req.body = body;
256+
return req;
257+
}
258+
259+
private static byte[] byteBufferToByteArray(final ByteBuffer byteBuffer) {
260+
final ByteBuffer duplicatedBuffer = byteBuffer.duplicate();
261+
final byte[] bytes = new byte[duplicatedBuffer.remaining()];
262+
duplicatedBuffer.get(bytes);
263+
return bytes;
264+
}
265+
266+
private static ByteBuffer serializeLegacyFileSealV2Body(
267+
final List<String> fileNames,
268+
final List<Long> fileLengths,
269+
final Map<String, String> parameters)
270+
throws IOException {
271+
try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
272+
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
273+
ReadWriteIOUtils.write(fileNames.size(), outputStream);
274+
for (final String fileName : fileNames) {
275+
ReadWriteIOUtils.write(fileName, outputStream);
276+
}
277+
ReadWriteIOUtils.write(fileLengths.size(), outputStream);
278+
for (final Long fileLength : fileLengths) {
279+
ReadWriteIOUtils.write(fileLength, outputStream);
280+
}
281+
ReadWriteIOUtils.write(parameters.size(), outputStream);
282+
for (final Map.Entry<String, String> entry : parameters.entrySet()) {
283+
ReadWriteIOUtils.write(entry.getKey(), outputStream);
284+
ReadWriteIOUtils.write(entry.getValue(), outputStream);
285+
}
286+
return ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
287+
}
288+
}
289+
290+
private static ByteBuffer serializeLegacyHandshakeV1Body(final String timestampPrecision)
291+
throws IOException {
292+
try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
293+
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
294+
ReadWriteIOUtils.write(timestampPrecision, outputStream);
295+
return ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
296+
}
297+
}
298+
299+
private static ByteBuffer serializeLegacyHandshakeV2Body(final Map<String, String> params)
300+
throws IOException {
301+
try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
302+
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
303+
ReadWriteIOUtils.write(params.size(), outputStream);
304+
for (final Map.Entry<String, String> entry : params.entrySet()) {
305+
ReadWriteIOUtils.write(entry.getKey(), outputStream);
306+
ReadWriteIOUtils.write(entry.getValue(), outputStream);
307+
}
308+
return ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
309+
}
310+
}
311+
312+
private static ByteBuffer serializeLegacyFilePieceBody(
313+
final String fileName, final long startWritingOffset, final byte[] filePiece)
314+
throws IOException {
315+
try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
316+
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
317+
ReadWriteIOUtils.write(fileName, outputStream);
318+
ReadWriteIOUtils.write(startWritingOffset, outputStream);
319+
ReadWriteIOUtils.write(new Binary(filePiece), outputStream);
320+
return ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
321+
}
322+
}
111323
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -641,15 +641,15 @@ private TSStatus loadSchemaSnapShot(
641641
PipeSchemaRegionSnapshotEvent.getStatementTypeSet(
642642
parameters.get(ColumnHeaderConstant.TYPE));
643643
final boolean isTreeModelDataAllowedToBeCaptured =
644-
parameters.containsKey(PipeTransferFileSealReqV2.TREE);
644+
PipeTransferFileSealReqV2.isTreeModelDataAllowedToBeCaptured(parameters);
645645
final TreePattern treePattern =
646646
TreePattern.parsePatternFromString(
647647
parameters.get(ColumnHeaderConstant.PATH_PATTERN),
648648
isTreeModelDataAllowedToBeCaptured,
649649
p -> new IoTDBTreePattern(isTreeModelDataAllowedToBeCaptured, p));
650650
final TablePattern tablePattern =
651651
new TablePattern(
652-
parameters.containsKey(PipeTransferFileSealReqV2.TABLE),
652+
PipeTransferFileSealReqV2.isTableModelDataAllowedToBeCaptured(parameters),
653653
parameters.get(PipeTransferFileSealReqV2.DATABASE_PATTERN),
654654
parameters.get(ColumnHeaderConstant.TABLE_NAME));
655655

0 commit comments

Comments
 (0)