jt2594838 commented on code in PR #18233:
URL: https://github.com/apache/iotdb/pull/18233#discussion_r3612341717
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReq.java:
##########
@@ -151,6 +152,13 @@ public static PipeTransferTabletBatchReq
toTPipeTransferReq(
return batchReq;
}
+ static int calculateSerializedSize(
+ final List<ByteBuffer> insertNodeBuffers, final List<ByteBuffer>
tabletBuffers) {
+ return Integer.BYTES * 3
Review Comment:
Add some comments to explain what the 3 is. The same below.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBinaryReqV2.java:
##########
@@ -162,6 +165,18 @@ public static byte[] toTPipeTransferBytes(final ByteBuffer
byteBuffer, final Str
}
}
+ private static int serializedStringSize(final String value) {
+ return Integer.BYTES + (value == null ? 0 :
value.getBytes(TSFileConfig.STRING_CHARSET).length);
+ }
Review Comment:
Move to a util.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReq.java:
##########
@@ -151,6 +152,13 @@ public static PipeTransferTabletBatchReq
toTPipeTransferReq(
return batchReq;
}
+ static int calculateSerializedSize(
+ final List<ByteBuffer> insertNodeBuffers, final List<ByteBuffer>
tabletBuffers) {
+ return Integer.BYTES * 3
+ + insertNodeBuffers.stream().mapToInt(ByteBuffer::limit).sum()
+ + tabletBuffers.stream().mapToInt(ByteBuffer::limit).sum();
+ }
Review Comment:
Better to use `limit - position`.
Where are the sizes of the buffers recorded? The sizes do not seem to be
counted in the size.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletInsertNodeReq.java:
##########
@@ -86,7 +86,14 @@ public static PipeTransferTabletInsertNodeReq
toTPipeTransferReq(final InsertNod
req.version = IoTDBSinkRequestVersion.VERSION_1.getVersion();
req.type = PipeRequestType.TRANSFER_TABLET_INSERT_NODE.getType();
- req.body = insertNode.serializeToByteBuffer();
+ try (final PublicBAOS byteArrayOutputStream =
+ new PublicBAOS(calculateSerializedSize(insertNode));
+ final DataOutputStream outputStream = new
DataOutputStream(byteArrayOutputStream)) {
+ insertNode.serialize(outputStream);
+ req.body = ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0,
byteArrayOutputStream.size());
+ } catch (final IOException e) {
+ throw new RuntimeException(e);
+ }
Review Comment:
May pass the size to `serializeToByteBuffer()`.
It would be even better to let the InsertNode handle this itself.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java:
##########
@@ -343,6 +343,75 @@ protected void serializeAttributes(DataOutputStream
stream) throws IOException {
subSerialize(stream);
}
+ @Override
+ protected int serializedAttributesSize() {
+ return PlanNodeType.BYTES + pipeSubSerializedSize();
+ }
+
+ /** Returns the exact size of the row fields written during Pipe
serialization. */
+ protected int pipeSubSerializedSize() {
+ return Long.BYTES
+ + ReadWriteIOUtils.sizeToWrite(targetPath.getFullPath())
+ + pipeMeasurementsAndValuesSerializedSize();
+ }
+
+ /** Returns the exact size of measurement and value fields written during
Pipe serialization. */
+ protected int pipeMeasurementsAndValuesSerializedSize() {
+ int size = Integer.BYTES + Byte.BYTES;
+
+ for (int i = 0; measurements != null && i < measurements.length; i++) {
+ if (!shouldSerializeMeasurement(i)) {
+ continue;
+ }
+ size +=
+ measurementSchemas == null
+ ? ReadWriteIOUtils.sizeToWrite(measurements[i])
+ : measurementSchemas[i].serializedSize();
+ }
+
+ for (int i = 0; values != null && i < values.length; i++) {
+ if (!shouldSerializeMeasurement(i)) {
+ continue;
+ }
+ size += pipeValueSerializedSize(i);
+ }
+
+ return size + Byte.BYTES + Byte.BYTES;
+ }
+
+ private int pipeValueSerializedSize(final int index) {
+ final TSDataType dataType = getDataTypeIfPresent(index);
+ if (values[index] == null) {
+ return Byte.BYTES + (dataType == null ? 0 : Byte.BYTES);
+ }
+
+ if (isNeedInferType) {
+ return Byte.BYTES +
ReadWriteIOUtils.sizeToWrite(values[index].toString());
+ }
+
+ switch (dataType) {
+ case BOOLEAN:
+ return Byte.BYTES + Byte.BYTES;
+ case INT32:
+ case DATE:
+ return Byte.BYTES + Integer.BYTES;
+ case INT64:
+ case TIMESTAMP:
+ return Byte.BYTES + Long.BYTES;
+ case FLOAT:
+ return Byte.BYTES + Float.BYTES;
+ case DOUBLE:
+ return Byte.BYTES + Double.BYTES;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ case OBJECT:
+ return Byte.BYTES + ReadWriteIOUtils.sizeToWrite((Binary)
values[index]);
+ default:
+ throw new UnSupportedDataTypeException(UNSUPPORTED_DATA_TYPE +
dataType);
+ }
Review Comment:
Or use org.apache.tsfile.enums.TSDataType#getDataTypeSize
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertMultiTabletsNode.java:
##########
@@ -280,6 +280,15 @@ protected void serializeAttributes(DataOutputStream
stream) throws IOException {
}
}
+ @Override
+ protected int serializedAttributesSize() {
+ int size = PlanNodeType.BYTES + Integer.BYTES;
+ for (final InsertTabletNode insertTabletNode : insertTabletNodeList) {
+ size += insertTabletNode.baseSubSerializedSizeForPipe();
+ }
+ return size + parentInsertTabletNodeIndexList.size() * Integer.BYTES;
+ }
Review Comment:
It feels weird that this method does not seem to be made for Pipe
exclusively, but it calls a method that seems to be.
Check the consistency of semantics and rename methods accordingly.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java:
##########
@@ -343,6 +343,75 @@ protected void serializeAttributes(DataOutputStream
stream) throws IOException {
subSerialize(stream);
}
+ @Override
+ protected int serializedAttributesSize() {
+ return PlanNodeType.BYTES + pipeSubSerializedSize();
+ }
+
+ /** Returns the exact size of the row fields written during Pipe
serialization. */
+ protected int pipeSubSerializedSize() {
+ return Long.BYTES
+ + ReadWriteIOUtils.sizeToWrite(targetPath.getFullPath())
+ + pipeMeasurementsAndValuesSerializedSize();
+ }
+
+ /** Returns the exact size of measurement and value fields written during
Pipe serialization. */
+ protected int pipeMeasurementsAndValuesSerializedSize() {
+ int size = Integer.BYTES + Byte.BYTES;
+
+ for (int i = 0; measurements != null && i < measurements.length; i++) {
+ if (!shouldSerializeMeasurement(i)) {
+ continue;
+ }
+ size +=
+ measurementSchemas == null
+ ? ReadWriteIOUtils.sizeToWrite(measurements[i])
+ : measurementSchemas[i].serializedSize();
+ }
+
+ for (int i = 0; values != null && i < values.length; i++) {
+ if (!shouldSerializeMeasurement(i)) {
+ continue;
+ }
+ size += pipeValueSerializedSize(i);
+ }
+
+ return size + Byte.BYTES + Byte.BYTES;
+ }
+
+ private int pipeValueSerializedSize(final int index) {
+ final TSDataType dataType = getDataTypeIfPresent(index);
+ if (values[index] == null) {
+ return Byte.BYTES + (dataType == null ? 0 : Byte.BYTES);
+ }
+
+ if (isNeedInferType) {
+ return Byte.BYTES +
ReadWriteIOUtils.sizeToWrite(values[index].toString());
+ }
+
+ switch (dataType) {
+ case BOOLEAN:
+ return Byte.BYTES + Byte.BYTES;
+ case INT32:
+ case DATE:
+ return Byte.BYTES + Integer.BYTES;
+ case INT64:
+ case TIMESTAMP:
+ return Byte.BYTES + Long.BYTES;
+ case FLOAT:
+ return Byte.BYTES + Float.BYTES;
+ case DOUBLE:
+ return Byte.BYTES + Double.BYTES;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ case OBJECT:
+ return Byte.BYTES + ReadWriteIOUtils.sizeToWrite((Binary)
values[index]);
+ default:
+ throw new UnSupportedDataTypeException(UNSUPPORTED_DATA_TYPE +
dataType);
+ }
Review Comment:
Use enhanced-switch and explicitly list all branches.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java:
##########
@@ -567,6 +567,97 @@ void subSerialize(DataOutputStream stream) throws
IOException {
ReadWriteIOUtils.write((byte) (isAligned ? 1 : 0), stream);
}
+ @Override
+ protected int serializedAttributesSize() {
+ return PlanNodeType.BYTES + baseSubSerializedSizeForPipe();
+ }
+
+ /**
+ * Returns the exact size written by {@link #subSerialize(DataOutputStream)}.
+ *
+ * <p>This deliberately excludes the plan-node type, id, and children. {@link
+ * InsertMultiTabletsNode} embeds tablet nodes by calling {@code
subSerialize}, rather than their
+ * complete plan-node serialization.
+ */
+ final int baseSubSerializedSizeForPipe() {
+ int size = ReadWriteIOUtils.sizeToWrite(targetPath.getFullPath());
+
+ size += Integer.BYTES;
+ size += Byte.BYTES;
Review Comment:
Add some comments for these. The same below.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java:
##########
Review Comment:
May consider caching the serialized size to avoid duplicated calculation
when serialization is called multiple times.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
