[IOTDB-2027] Rollback invalid entry after wal writing failure (#4424)

Author: HeimingZ

server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java

@@ -27,7 +27,6 @@
 import org.apache.iotdb.db.qp.executor.PlanExecutor;
 import org.apache.iotdb.db.qp.physical.PhysicalPlan;
 import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
 
 import io.moquette.interception.AbstractInterceptHandler;
 import io.moquette.interception.messages.InterceptPublishMessage;
@@ -93,16 +92,15 @@ public void onPublish(InterceptPublishMessage msg) {
         continue;
       }
 
-      InsertRowPlan plan = new InsertRowPlan();
-      plan.setTime(event.getTimestamp());
-      plan.setMeasurements(event.getMeasurements().toArray(new String[0]));
-      plan.setValues(event.getValues().toArray(new Object[0]));
-      plan.setDataTypes(new TSDataType[event.getValues().size()]);
-      plan.setNeedInferType(true);
-
       boolean status = false;
       try {
-        plan.setDeviceId(new PartialPath(event.getDevice()));
+        PartialPath path = new PartialPath(event.getDevice());
+        InsertRowPlan plan =
+            new InsertRowPlan(
+                path,
+                event.getTimestamp(),
+                event.getMeasurements().toArray(new String[0]),
+                event.getValues().toArray(new String[0]));
         status = executeNonQuery(plan);
       } catch (Exception e) {
         LOG.warn(

server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java

@@ -73,6 +73,9 @@
 import org.apache.iotdb.db.qp.utils.EmptyOutputStream;
 import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -82,6 +85,7 @@
 
 /** This class is a abstract class for all type of PhysicalPlan. */
 public abstract class PhysicalPlan {
+  private static final Logger logger = LoggerFactory.getLogger(PhysicalPlan.class);
 
   private static final String SERIALIZATION_UNIMPLEMENTED = "serialization unimplemented";
 
@@ -182,11 +186,27 @@ public void serialize(DataOutputStream stream) throws IOException {
 
   /**
    * Serialize the plan into the given buffer. This is provided for WAL, so fields that can be
-   * recovered will not be serialized.
+   * recovered will not be serialized. If error occurs when serializing this plan, the buffer will
+   * be reset.
    *
    * @param buffer
    */
   public void serialize(ByteBuffer buffer) {
+    buffer.mark();
+    try {
+      serializeImpl(buffer);
+    } catch (UnsupportedOperationException e) {
+      // ignore and throw
+      throw e;
+    } catch (Exception e) {
+      logger.error(
+          "Rollback buffer entry because error occurs when serializing this physical plan.", e);
+      buffer.reset();
+      throw e;
+    }
+  }
+
+  protected void serializeImpl(ByteBuffer buffer) {
     throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
   }
 

server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java

@@ -151,7 +151,7 @@ public void serialize(DataOutputStream stream) throws IOException {
   }
 
   @Override
-  public void serialize(ByteBuffer buffer) {
+  public void serializeImpl(ByteBuffer buffer) {
     int type = PhysicalPlanType.DELETE.ordinal();
     buffer.put((byte) type);
     buffer.putLong(deleteStartTime);

server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertMultiTabletPlan.java

@@ -258,7 +258,7 @@ public void recoverFromFailure() {
   }
 
   @Override
-  public void serialize(ByteBuffer buffer) {
+  public void serializeImpl(ByteBuffer buffer) {
     int type = PhysicalPlanType.MULTI_BATCH_INSERT.ordinal();
     buffer.put((byte) type);
     buffer.putInt(insertTabletPlanList.size());

server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java

@@ -478,7 +478,7 @@ public void fillValues(ByteBuffer buffer) throws QueryProcessException {
   }
 
   @Override
-  public void serialize(ByteBuffer buffer) {
+  public void serializeImpl(ByteBuffer buffer) {
     int type = PhysicalPlanType.INSERT.ordinal();
     buffer.put((byte) type);
     subSerialize(buffer);

server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowsOfOneDevicePlan.java

@@ -159,7 +159,7 @@ public void serialize(DataOutputStream stream) throws IOException {
   }
 
   @Override
-  public void serialize(ByteBuffer buffer) {
+  public void serializeImpl(ByteBuffer buffer) {
     int type = PhysicalPlanType.BATCH_INSERT_ONE_DEVICE.ordinal();
     buffer.put((byte) type);
 

server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowsPlan.java

@@ -168,7 +168,7 @@ public int hashCode() {
   }
 
   @Override
-  public void serialize(ByteBuffer buffer) {
+  public void serializeImpl(ByteBuffer buffer) {
     int type = PhysicalPlanType.BATCH_INSERT_ROWS.ordinal();
     buffer.put((byte) type);
     buffer.putInt(insertRowPlanList.size());

server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java

@@ -247,7 +247,7 @@ private void writeValues(DataOutputStream stream) throws IOException {
   }
 
   @Override
-  public void serialize(ByteBuffer buffer) {
+  public void serializeImpl(ByteBuffer buffer) {
     int type = PhysicalPlanType.BATCHINSERT.ordinal();
     buffer.put((byte) type);
     subSerialize(buffer);

server/src/main/java/org/apache/iotdb/db/qp/physical/crud/SelectIntoPlan.java

@@ -68,7 +68,7 @@ public void serialize(DataOutputStream outputStream) throws IOException {
   }
 
   @Override
-  public void serialize(ByteBuffer buffer) {
+  public void serializeImpl(ByteBuffer buffer) {
     buffer.put((byte) PhysicalPlanType.SELECT_INTO.ordinal());
 
     queryPlan.serialize(buffer);

server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ActivateTemplatePlan.java

@@ -57,7 +57,7 @@ public PartialPath getPrefixPath() {
   }
 
   @Override
-  public void serialize(ByteBuffer buffer) {
+  public void serializeImpl(ByteBuffer buffer) {
     buffer.put((byte) PhysicalPlanType.ACTIVATE_TEMPLATE.ordinal());
     ReadWriteIOUtils.write(prefixPath.getFullPath(), buffer);
     buffer.putLong(index);