Pipe: Convert data types when data sync sink metadata does not match integration test

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBDataRegionAsyncConnector.java

@@ -518,7 +518,9 @@ public synchronized void clearRetryEventsReferenceCount() {
 
   @Override
   public synchronized void discardEventsOfPipe(final String pipeNameToDrop) {
-    tabletBatchBuilder.discardEventsOfPipe(pipeNameToDrop);
+    if (isTabletBatchModeEnabled) {
+      tabletBatchBuilder.discardEventsOfPipe(pipeNameToDrop);
+    }
     retryEventQueue.removeIf(
         event -> {
           if (event instanceof EnrichedEvent

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/converter/ValueConverter.java

@@ -26,8 +26,12 @@
 import org.apache.tsfile.common.conf.TSFileConfig;
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.DateUtils;
 
-import java.time.ZoneId;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
 
 public class ValueConverter {
 
@@ -299,6 +303,11 @@ public static Object convert(
 
   private static final Binary BINARY_TRUE = parseString(Boolean.TRUE.toString());
   private static final Binary BINARY_FALSE = parseString(Boolean.FALSE.toString());
+  private static final int TRUE_DATE = DateUtils.parseDateExpressionToInt(LocalDate.of(1970, 1, 2));
+  private static final int FALSE_DATE =
+      DateUtils.parseDateExpressionToInt(LocalDate.of(1970, 1, 1));
+  private static final int DEFAULT_DATE =
+      DateUtils.parseDateExpressionToInt(LocalDate.of(1970, 1, 1));
 
   public static int convertBooleanToInt32(final boolean value) {
     return value ? 1 : 0;
@@ -325,7 +334,7 @@ public static long convertBooleanToTimestamp(final boolean value) {
   }
 
   public static int convertBooleanToDate(final boolean value) {
-    return value ? 1 : 0;
+    return value ? TRUE_DATE : FALSE_DATE;
   }
 
   public static Binary convertBooleanToBlob(final boolean value) {
@@ -363,7 +372,12 @@ public static long convertInt32ToTimestamp(final int value) {
   }
 
   public static int convertInt32ToDate(final int value) {
-    return value;
+    try {
+      DateUtils.parseIntToLocalDate(value);
+      return value;
+    } catch (Exception e) {
+      return DEFAULT_DATE;
+    }
   }
 
   public static Binary convertInt32ToBlob(final int value) {
@@ -401,7 +415,13 @@ public static long convertInt64ToTimestamp(final long value) {
   }
 
   public static int convertInt64ToDate(final long value) {
-    return (int) value;
+    try {
+      int data = (int) value;
+      DateUtils.parseIntToLocalDate(data);
+      return data;
+    } catch (Exception e) {
+      return DEFAULT_DATE;
+    }
   }
 
   public static Binary convertInt64ToBlob(final long value) {
@@ -439,7 +459,13 @@ public static long convertFloatToTimestamp(final float value) {
   }
 
   public static int convertFloatToDate(final float value) {
-    return (int) value;
+    try {
+      int data = (int) value;
+      DateUtils.parseIntToLocalDate(data);
+      return data;
+    } catch (Exception e) {
+      return DEFAULT_DATE;
+    }
   }
 
   public static Binary convertFloatToBlob(final float value) {
@@ -477,7 +503,13 @@ public static long convertDoubleToTimestamp(final double value) {
   }
 
   public static int convertDoubleToDate(final double value) {
-    return (int) value;
+    try {
+      int data = (int) value;
+      DateUtils.parseIntToLocalDate(data);
+      return data;
+    } catch (Exception e) {
+      return DEFAULT_DATE;
+    }
   }
 
   public static Binary convertDoubleToBlob(final double value) {
@@ -553,7 +585,12 @@ public static Binary convertTimestampToText(final long value) {
   }
 
   public static int convertTimestampToDate(final long value) {
-    return (int) value;
+    try {
+      Instant instant = Instant.ofEpochMilli(value);
+      return DateUtils.parseDateExpressionToInt(instant.atZone(ZoneOffset.UTC).toLocalDate());
+    } catch (Exception e) {
+      return DEFAULT_DATE;
+    }
   }
 
   public static Binary convertTimestampToBlob(final long value) {
@@ -567,7 +604,7 @@ public static Binary convertTimestampToString(final long value) {
   ///////////// DATE //////////////
 
   public static boolean convertDateToBoolean(final int value) {
-    return value != 0;
+    return value != FALSE_DATE;
   }
 
   public static int convertDateToInt32(final int value) {
@@ -591,7 +628,14 @@ public static Binary convertDateToText(final int value) {
   }
 
   public static long convertDateToTimestamp(final int value) {
-    return value;
+    try {
+      LocalDate date = DateUtils.parseIntToLocalDate(value);
+      ZonedDateTime dateTime = date.atStartOfDay(ZoneOffset.UTC);
+      Instant instant = dateTime.toInstant();
+      return instant.toEpochMilli();
+    } catch (Exception e) {
+      return 0L;
+    }
   }
 
   public static Binary convertDateToBlob(final int value) {
@@ -753,23 +797,25 @@ private static long parseTimestamp(final String value) {
     try {
       return TypeInferenceUtils.isNumber(value)
           ? Long.parseLong(value)
-          : DateTimeUtils.parseDateTimeExpressionToLong(
-              StringUtils.trim(value), ZoneId.systemDefault());
+          : DateTimeUtils.parseDateTimeExpressionToLong(StringUtils.trim(value), ZoneOffset.UTC);
     } catch (final Exception e) {
       return 0L;
     }
   }
 
   private static int parseDate(final String value) {
     if (value == null || value.isEmpty()) {
-      return 0;
+      return DEFAULT_DATE;
     }
     try {
-      return TypeInferenceUtils.isNumber(value)
-          ? Integer.parseInt(value)
-          : DateTimeUtils.parseDateExpressionToInt(StringUtils.trim(value));
+      if (TypeInferenceUtils.isNumber(value)) {
+        int date = Integer.parseInt(value);
+        DateUtils.parseIntToLocalDate(date);
+        return date;
+      }
+      return DateTimeUtils.parseDateExpressionToInt(StringUtils.trim(value));
     } catch (final Exception e) {
-      return 0;
+      return DEFAULT_DATE;
     }
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java

@@ -42,6 +42,7 @@ public PipeConvertedInsertRowStatement(final InsertRowStatement insertRowStateme
     // Statement
     isDebug = insertRowStatement.isDebug();
     // InsertBaseStatement
+    insertRowStatement.removeAllFailedMeasurementMarks();
     devicePath = insertRowStatement.getDevicePath();
     isAligned = insertRowStatement.isAligned();
     measurementSchemas = insertRowStatement.getMeasurementSchemas();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertTabletStatement.java

@@ -36,6 +36,7 @@ public PipeConvertedInsertTabletStatement(final InsertTabletStatement insertTabl
     // Statement
     isDebug = insertTabletStatement.isDebug();
     // InsertBaseStatement
+    insertTabletStatement.removeAllFailedMeasurementMarks();
     devicePath = insertTabletStatement.getDevicePath();
     isAligned = insertTabletStatement.isAligned();
     measurementSchemas = insertTabletStatement.getMeasurementSchemas();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertBaseStatement.java

@@ -264,6 +264,11 @@ public void markFailedMeasurement(int index, Exception cause) {
     throw new UnsupportedOperationException();
   }
 
+  /** * Resets the state of all measurements marked as failed, clearing the failure records. */
+  public void removeAllFailedMeasurementMarks() {
+    throw new UnsupportedOperationException();
+  }
+
   public boolean hasValidMeasurements() {
     for (Object o : measurements) {
       if (o != null) {
@@ -372,6 +377,18 @@ public FailedMeasurementInfo(
       this.value = value;
       this.cause = cause;
     }
+
+    public String getMeasurement() {
+      return measurement;
+    }
+
+    public TSDataType getDataType() {
+      return dataType;
+    }
+
+    public Object getValue() {
+      return value;
+    }
   }
 
   // endregion

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowStatement.java

@@ -276,6 +276,20 @@ public void markFailedMeasurement(int index, Exception cause) {
     values[index] = null;
   }
 
+  @Override
+  public void removeAllFailedMeasurementMarks() {
+    if (failedMeasurementIndex2Info == null) {
+      return;
+    }
+    failedMeasurementIndex2Info.forEach(
+        (index, info) -> {
+          measurements[index] = info.getMeasurement();
+          dataTypes[index] = info.getDataType();
+          values[index] = info.getValue();
+        });
+    failedMeasurementIndex2Info.clear();
+  }
+
   @Override
   public void semanticCheck() {
     super.semanticCheck();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertTabletStatement.java

@@ -228,6 +228,20 @@ public void markFailedMeasurement(int index, Exception cause) {
     columns[index] = null;
   }
 
+  @Override
+  public void removeAllFailedMeasurementMarks() {
+    if (failedMeasurementIndex2Info == null) {
+      return;
+    }
+    failedMeasurementIndex2Info.forEach(
+        (index, info) -> {
+          measurements[index] = info.getMeasurement();
+          dataTypes[index] = info.getDataType();
+          columns[index] = info.getValue();
+        });
+    failedMeasurementIndex2Info.clear();
+  }
+
   @Override
   public void semanticCheck() {
     super.semanticCheck();