Fix legacy raw tablet fallback

Author: Caideyipi

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

@@ -30,6 +30,7 @@
 import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
 import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
 
+import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.utils.PublicBAOS;
 import org.apache.tsfile.utils.ReadWriteIOUtils;
 import org.apache.tsfile.write.record.Tablet;
@@ -164,6 +165,9 @@ private void deserializeTPipeTransferRawReq(
       final InsertTabletStatement insertTabletStatement =
           TabletStatementConverter.deserializeStatementFromTabletFormat(
               buffer, false, tabletStringInternPool);
+      // Legacy tablets do not serialize column categories. Since hasSchema=1 can be
+      // misread as FIELD, the current reader may return a corrupt statement instead of failing.
+      ensureStatementDeserializedFromCurrentTabletFormat(insertTabletStatement);
       isAligned = insertTabletStatement.isAligned();
       statement = insertTabletStatement;
       return;
@@ -188,6 +192,44 @@ private void deserializeTPipeTransferRawReq(
     isAligned = ReadWriteIOUtils.readBool(buffer);
   }
 
+  private static void ensureStatementDeserializedFromCurrentTabletFormat(
+      final InsertTabletStatement statement) {
+    final String[] measurements = statement.getMeasurements();
+    final TSDataType[] dataTypes = statement.getDataTypes();
+
+    if (Objects.isNull(measurements)
+        || Objects.isNull(dataTypes)
+        || measurements.length != dataTypes.length) {
+      throw new IllegalArgumentException(
+          "Incomplete schema in current tablet format deserialization.");
+    }
+
+    final Object[] columns = statement.getColumns();
+    if (Objects.nonNull(columns) && columns.length != measurements.length) {
+      throw new IllegalArgumentException(
+          "Column count is inconsistent with schema count in current tablet format deserialization.");
+    }
+
+    for (int i = 0; i < measurements.length; ++i) {
+      if (Objects.isNull(measurements[i]) || Objects.isNull(dataTypes[i])) {
+        throw new IllegalArgumentException(
+            "Incomplete measurement schema in current tablet format deserialization.");
+      }
+      if (statement.getRowCount() > 0 && (Objects.isNull(columns) || Objects.isNull(columns[i]))) {
+        throw new IllegalArgumentException(
+            "Incomplete column values in current tablet format deserialization.");
+      }
+    }
+
+    final long[] times = statement.getTimes();
+    if (statement.getRowCount() > 0
+        && measurements.length > 0
+        && (Objects.isNull(times) || times.length < statement.getRowCount())) {
+      throw new IllegalArgumentException(
+          "Incomplete timestamps in current tablet format deserialization.");
+    }
+  }
+
   /////////////////////////////// Air Gap ///////////////////////////////
 
   /**