Pipe: Modify the TableRawReq deserialization method to support directconversion to TableStatement. (#16844)

* Pipe: Modify the TableRawReq deserialization method to support direct conversion to TableStatement.

* fix

* fix

* fix

* fix

* fix

* update

* update

* update

* refactor: optimize TabletStatementConverter according to code review

- Optimize times array copy: skip copy when lengths are equal, use System.arraycopy
- Add warning logs when times array is null or too small
- Ensure all arrays (values, times, bitMaps) are copied to rowSize length for immutability
- Filter out null columns when converting Statement to Tablet
- Rename idColumnIndices to tagColumnIndices
- Add skipString method to avoid constructing temporary objects
- Add comments explaining skipped fields in readMeasurement
- Use direct buffer position increment instead of reading bytes for skipping
- Ensure all column values are copied to ensure immutability

* update

* update

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2.java

@@ -33,7 +33,6 @@
 
 import org.apache.tsfile.utils.PublicBAOS;
 import org.apache.tsfile.utils.ReadWriteIOUtils;
-import org.apache.tsfile.write.record.Tablet;
 
 import java.io.DataOutputStream;
 import java.io.IOException;
@@ -247,11 +246,7 @@ public static PipeTransferTabletBatchReqV2 fromTPipeTransferReq(
 
     size = ReadWriteIOUtils.readInt(transferReq.body);
     for (int i = 0; i < size; ++i) {
-      batchReq.tabletReqs.add(
-          PipeTransferTabletRawReqV2.toTPipeTransferRawReq(
-              Tablet.deserialize(transferReq.body),
-              ReadWriteIOUtils.readBool(transferReq.body),
-              ReadWriteIOUtils.readString(transferReq.body)));
+      batchReq.tabletReqs.add(PipeTransferTabletRawReqV2.toTPipeTransferRawReq(transferReq.body));
     }
 
     batchReq.version = transferReq.version;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReq.java

@@ -22,6 +22,7 @@
 import org.apache.iotdb.commons.exception.MetadataException;
 import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.IoTDBSinkRequestVersion;
 import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType;
+import org.apache.iotdb.db.pipe.sink.util.TabletStatementConverter;
 import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTreeModelTabletEventSorter;
 import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
 import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
@@ -43,27 +44,55 @@ public class PipeTransferTabletRawReq extends TPipeTransferReq {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(PipeTransferTabletRawReq.class);
 
-  protected transient Tablet tablet;
+  protected transient InsertTabletStatement statement;
+
   protected transient boolean isAligned;
+  protected transient Tablet tablet;
 
+  /**
+   * Get Tablet. If tablet is null, convert from statement.
+   *
+   * @return Tablet object
+   */
   public Tablet getTablet() {
+    if (tablet == null && statement != null) {
+      try {
+        tablet = statement.convertToTablet();
+      } catch (final MetadataException e) {
+        LOGGER.warn("Failed to convert statement to tablet.", e);
+        return null;
+      }
+    }
     return tablet;
   }
 
   public boolean getIsAligned() {
     return isAligned;
   }
 
+  /**
+   * Construct Statement. If statement already exists, return it. Otherwise, convert from tablet.
+   *
+   * @return InsertTabletStatement
+   */
   public InsertTabletStatement constructStatement() {
+    if (statement != null) {
+      new PipeTreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary();
+      return statement;
+    }
+
+    // Sort and deduplicate tablet before converting
     new PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
 
     try {
       if (isTabletEmpty(tablet)) {
         // Empty statement, will be filtered after construction
-        return new InsertTabletStatement();
+        statement = new InsertTabletStatement();
+        return statement;
       }
 
-      return new InsertTabletStatement(tablet, isAligned, null);
+      statement = new InsertTabletStatement(tablet, isAligned, null);
+      return statement;
     } catch (final MetadataException e) {
       LOGGER.warn("Generate Statement from tablet {} error.", tablet, e);
       return null;
@@ -107,8 +136,20 @@ public static PipeTransferTabletRawReq toTPipeTransferReq(
   public static PipeTransferTabletRawReq fromTPipeTransferReq(final TPipeTransferReq transferReq) {
     final PipeTransferTabletRawReq tabletReq = new PipeTransferTabletRawReq();
 
-    tabletReq.tablet = Tablet.deserialize(transferReq.body);
-    tabletReq.isAligned = ReadWriteIOUtils.readBool(transferReq.body);
+    final ByteBuffer buffer = transferReq.body;
+    final int startPosition = buffer.position();
+    try {
+      // V1: no databaseName, readDatabaseName = false
+      final InsertTabletStatement insertTabletStatement =
+          TabletStatementConverter.deserializeStatementFromTabletFormat(buffer, false);
+      tabletReq.isAligned = insertTabletStatement.isAligned();
+      // devicePath is already set in deserializeStatementFromTabletFormat for V1 format
+      tabletReq.statement = insertTabletStatement;
+    } catch (final Exception e) {
+      buffer.position(startPosition);
+      tabletReq.tablet = Tablet.deserialize(buffer);
+      tabletReq.isAligned = ReadWriteIOUtils.readBool(buffer);
+    }
 
     tabletReq.version = transferReq.version;
     tabletReq.type = transferReq.type;
@@ -118,18 +159,56 @@ public static PipeTransferTabletRawReq fromTPipeTransferReq(final TPipeTransferR
 
   /////////////////////////////// Air Gap ///////////////////////////////
 
-  public static byte[] toTPipeTransferBytes(final Tablet tablet, final boolean isAligned)
-      throws IOException {
+  /**
+   * Serialize to bytes. If tablet is null, convert from statement first.
+   *
+   * @return serialized bytes
+   * @throws IOException if serialization fails
+   */
+  public byte[] toTPipeTransferBytes() throws IOException {
+    Tablet tabletToSerialize = tablet;
+    boolean isAlignedToSerialize = isAligned;
+
+    // If tablet is null, convert from statement
+    if (tabletToSerialize == null && statement != null) {
+      try {
+        tabletToSerialize = statement.convertToTablet();
+        isAlignedToSerialize = statement.isAligned();
+      } catch (final MetadataException e) {
+        throw new IOException("Failed to convert statement to tablet for serialization", e);
+      }
+    }
+
+    if (tabletToSerialize == null) {
+      throw new IOException("Cannot serialize: both tablet and statement are null");
+    }
+
     try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
         final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
       ReadWriteIOUtils.write(IoTDBSinkRequestVersion.VERSION_1.getVersion(), outputStream);
       ReadWriteIOUtils.write(PipeRequestType.TRANSFER_TABLET_RAW.getType(), outputStream);
-      tablet.serialize(outputStream);
-      ReadWriteIOUtils.write(isAligned, outputStream);
+      tabletToSerialize.serialize(outputStream);
+      ReadWriteIOUtils.write(isAlignedToSerialize, outputStream);
       return byteArrayOutputStream.toByteArray();
     }
   }
 
+  /**
+   * Static method for backward compatibility. Creates a temporary instance and serializes.
+   *
+   * @param tablet Tablet to serialize
+   * @param isAligned whether aligned
+   * @return serialized bytes
+   * @throws IOException if serialization fails
+   */
+  public static byte[] toTPipeTransferBytes(final Tablet tablet, final boolean isAligned)
+      throws IOException {
+    final PipeTransferTabletRawReq req = new PipeTransferTabletRawReq();
+    req.tablet = tablet;
+    req.isAligned = isAligned;
+    return req.toTPipeTransferBytes();
+  }
+
   /////////////////////////////// Object ///////////////////////////////
 
   @Override
@@ -141,7 +220,16 @@ public boolean equals(final Object obj) {
       return false;
     }
     final PipeTransferTabletRawReq that = (PipeTransferTabletRawReq) obj;
-    return Objects.equals(tablet, that.tablet)
+    // Compare statement if both have it, otherwise compare tablet
+    if (statement != null && that.statement != null) {
+      return Objects.equals(statement, that.statement)
+          && isAligned == that.isAligned
+          && version == that.version
+          && type == that.type
+          && Objects.equals(body, that.body);
+    }
+    // Fallback to tablet comparison
+    return Objects.equals(getTablet(), that.getTablet())
         && isAligned == that.isAligned
         && version == that.version
         && type == that.type
@@ -150,6 +238,6 @@ public boolean equals(final Object obj) {
 
   @Override
   public int hashCode() {
-    return Objects.hash(tablet, isAligned, version, type, body);
+    return Objects.hash(getTablet(), isAligned, version, type, body);
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReqV2.java

@@ -22,6 +22,7 @@
 import org.apache.iotdb.commons.exception.MetadataException;
 import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.IoTDBSinkRequestVersion;
 import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType;
+import org.apache.iotdb.db.pipe.sink.util.TabletStatementConverter;
 import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTableModelTabletEventSorter;
 import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTreeModelTabletEventSorter;
 import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
@@ -52,6 +53,16 @@ public String getDataBaseName() {
 
   @Override
   public InsertTabletStatement constructStatement() {
+    if (statement != null) {
+      if (Objects.isNull(dataBaseName)) {
+        new PipeTreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary();
+      } else {
+        new PipeTableModelTabletEventSorter(statement).sortByTimestampIfNecessary();
+      }
+
+      return statement;
+    }
+
     if (Objects.isNull(dataBaseName)) {
       new PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
     } else {
@@ -86,6 +97,16 @@ public static PipeTransferTabletRawReqV2 toTPipeTransferRawReq(
     return tabletReq;
   }
 
+  public static PipeTransferTabletRawReqV2 toTPipeTransferRawReq(final ByteBuffer buffer) {
+    final PipeTransferTabletRawReqV2 tabletReq = new PipeTransferTabletRawReqV2();
+
+    tabletReq.deserializeTPipeTransferRawReq(buffer);
+    tabletReq.version = IoTDBSinkRequestVersion.VERSION_1.getVersion();
+    tabletReq.type = PipeRequestType.TRANSFER_TABLET_RAW_V2.getType();
+
+    return tabletReq;
+  }
+
   /////////////////////////////// Thrift ///////////////////////////////
 
   public static PipeTransferTabletRawReqV2 toTPipeTransferReq(
@@ -114,13 +135,11 @@ public static PipeTransferTabletRawReqV2 fromTPipeTransferReq(
       final TPipeTransferReq transferReq) {
     final PipeTransferTabletRawReqV2 tabletReq = new PipeTransferTabletRawReqV2();
 
-    tabletReq.tablet = Tablet.deserialize(transferReq.body);
-    tabletReq.isAligned = ReadWriteIOUtils.readBool(transferReq.body);
-    tabletReq.dataBaseName = ReadWriteIOUtils.readString(transferReq.body);
+    tabletReq.deserializeTPipeTransferRawReq(transferReq.body);
+    tabletReq.body = transferReq.body;
 
     tabletReq.version = transferReq.version;
     tabletReq.type = transferReq.type;
-    tabletReq.body = transferReq.body;
 
     return tabletReq;
   }
@@ -161,4 +180,27 @@ public boolean equals(Object o) {
   public int hashCode() {
     return Objects.hash(super.hashCode(), dataBaseName);
   }
+
+  /////////////////////////////// Util ///////////////////////////////
+
+  public void deserializeTPipeTransferRawReq(final ByteBuffer buffer) {
+    final int startPosition = buffer.position();
+    try {
+      // V2: read databaseName, readDatabaseName = true
+      final InsertTabletStatement insertTabletStatement =
+          TabletStatementConverter.deserializeStatementFromTabletFormat(buffer, true);
+      this.isAligned = insertTabletStatement.isAligned();
+      // databaseName is already set in deserializeStatementFromTabletFormat when
+      // readDatabaseName=true
+      this.dataBaseName = insertTabletStatement.getDatabaseName().orElse(null);
+      this.statement = insertTabletStatement;
+    } catch (final Exception e) {
+      // If Statement deserialization fails, fallback to Tablet format
+      // Reset buffer position for Tablet deserialization
+      buffer.position(startPosition);
+      this.tablet = Tablet.deserialize(buffer);
+      this.isAligned = ReadWriteIOUtils.readBool(buffer);
+      this.dataBaseName = ReadWriteIOUtils.readString(buffer);
+    }
+  }
 }