Subscription: Fix DATE serialization for consensus dataset

integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionDataTableIT.java

@@ -20,20 +20,29 @@
 package org.apache.iotdb.subscription.it.consensus.local.tablemodel;
 
 import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionDataSet;
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
 import org.apache.iotdb.itbase.category.LocalStandaloneIT;
 import org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumer;
+import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
 import org.apache.iotdb.subscription.it.consensus.local.AbstractSubscriptionConsensusLocalIT;
 
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.write.record.Tablet;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 import org.junit.runner.RunWith;
 
 import java.time.Duration;
+import java.time.LocalDate;
 import java.util.Arrays;
 import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Locale;
 import java.util.Set;
 
@@ -51,6 +60,55 @@ public class IoTDBConsensusSubscriptionDataTableIT extends AbstractSubscriptionC
           + "s_bool BOOLEAN FIELD, "
           + "s_text TEXT FIELD";
 
+  @Test
+  public void testDateFieldInsertedByTablet() throws Exception {
+    final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
+        ConsensusSubscriptionTableITSupport.newIdentifiers("table_data_date_tablet");
+    final String database = ids.getDatabase();
+    final String tableName = "table_0";
+    final LocalDate expectedDate = LocalDate.of(2026, 1, 31);
+    SubscriptionTablePullConsumer consumer = null;
+
+    try {
+      ConsensusSubscriptionTableITSupport.createDatabaseAndTable(
+          database, tableName, "device_id STRING TAG, value DATE FIELD");
+      ConsensusSubscriptionTableITSupport.createConsensusTopic(ids.getTopic(), database, tableName);
+      consumer =
+          ConsensusSubscriptionTableITSupport.createConsumer(
+              ids.getConsumerId(), ids.getConsumerGroupId());
+      consumer.subscribe(ids.getTopic());
+      ConsensusSubscriptionTableITSupport.pause(1_000L);
+
+      try (final ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) {
+        session.executeNonQueryStatement("use " + database);
+        final Tablet tablet =
+            new Tablet(
+                tableName,
+                Arrays.asList("device_id", "value"),
+                Arrays.asList(TSDataType.STRING, TSDataType.DATE),
+                Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD),
+                1);
+        tablet.addTimestamp(0, 1L);
+        tablet.addValue("device_id", 0, "device_0");
+        tablet.addValue("value", 0, expectedDate);
+        session.insert(tablet);
+        session.executeNonQueryStatement("flush");
+
+        try (final SessionDataSet dataSet =
+            session.executeQueryStatement("select count(*) from " + tableName)) {
+          Assert.assertTrue(dataSet.hasNext());
+          Assert.assertEquals(1L, dataSet.next().getFields().get(0).getLongV());
+          Assert.assertFalse(dataSet.hasNext());
+        }
+      }
+
+      assertDateFieldConsumed(consumer, database, tableName, expectedDate);
+      ConsensusSubscriptionTableITSupport.assertNoMoreMessages(consumer, 3, Duration.ofMillis(500));
+    } finally {
+      ConsensusSubscriptionTableITSupport.cleanup(consumer, ids.getTopic(), database);
+    }
+  }
+
   @Test
   public void testTypedRowsAcrossTimePartitions() throws Exception {
     final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
@@ -134,4 +192,60 @@ public void testTypedRowsAcrossTimePartitions() throws Exception {
       ConsensusSubscriptionTableITSupport.cleanup(consumer, ids.getTopic(), database);
     }
   }
+
+  private static void assertDateFieldConsumed(
+      final SubscriptionTablePullConsumer consumer,
+      final String database,
+      final String tableName,
+      final LocalDate expectedDate)
+      throws Exception {
+    for (int round = 0; round < 60; round++) {
+      final List<SubscriptionMessage> messages = consumer.poll(Duration.ofSeconds(1));
+      if (messages.isEmpty()) {
+        continue;
+      }
+
+      try {
+        int matchedRows = 0;
+        for (final SubscriptionMessage message : messages) {
+          for (final ResultSet resultSet : message.getResultSets()) {
+            final SubscriptionRecordHandler.SubscriptionResultSet subscriptionResultSet =
+                (SubscriptionRecordHandler.SubscriptionResultSet) resultSet;
+            if (!database.equals(subscriptionResultSet.getDatabaseName())
+                || !tableName.equals(subscriptionResultSet.getTableName())) {
+              continue;
+            }
+
+            final Tablet tablet = subscriptionResultSet.getTablet();
+            int dateColumnIndex = -1;
+            for (int columnIndex = 0; columnIndex < tablet.getSchemas().size(); columnIndex++) {
+              if ("value".equals(tablet.getSchemas().get(columnIndex).getMeasurementName())) {
+                dateColumnIndex = columnIndex;
+                break;
+              }
+            }
+            Assert.assertTrue(
+                "DATE field is absent from the consumed tablet", dateColumnIndex >= 0);
+            Assert.assertEquals(
+                TSDataType.DATE, tablet.getSchemas().get(dateColumnIndex).getType());
+
+            for (int rowIndex = 0; rowIndex < tablet.getRowSize(); rowIndex++) {
+              if (tablet.getTimestamp(rowIndex) == 1L) {
+                Assert.assertEquals(expectedDate, tablet.getValue(rowIndex, dateColumnIndex));
+                matchedRows++;
+              }
+            }
+          }
+        }
+        if (matchedRows > 0) {
+          Assert.assertEquals(1, matchedRows);
+          return;
+        }
+      } finally {
+        consumer.commitSync(messages);
+      }
+    }
+
+    Assert.fail("DATE tablet row was not consumed within 60 poll rounds");
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java

@@ -48,12 +48,14 @@
 import org.apache.tsfile.file.metadata.IDeviceID;
 import org.apache.tsfile.utils.Binary;
 import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.utils.DateUtils;
 import org.apache.tsfile.write.record.Tablet;
 import org.apache.tsfile.write.schema.IMeasurementSchema;
 import org.apache.tsfile.write.schema.MeasurementSchema;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.LocalDate;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -327,7 +329,11 @@ private List<Tablet> convertInsertTabletNode(final InsertTabletNode node) {
 
     for (int i = 0; i < columnCount; i++) {
       final int originalColIdx = allColumnsMatch ? i : matchedColumnIndices.get(i);
-      newColumns[i] = columns[originalColIdx];
+      newColumns[i] =
+          convertColumnToTabletRepresentation(
+              dataTypes[originalColIdx],
+              columns[originalColIdx],
+              bitMaps != null ? bitMaps[originalColIdx] : null);
       if (bitMaps != null && bitMaps[originalColIdx] != null) {
         newBitMaps[i] = bitMaps[originalColIdx];
       }
@@ -535,7 +541,11 @@ private List<Tablet> convertRelationalInsertTabletNode(final RelationalInsertTab
       }
       schemas.add(new MeasurementSchema(measurements[i], dataTypes[i]));
       columnTypes.add(toTsFileColumnCategory(node.getColumnCategories(), i));
-      validColumns.add(columns[i]);
+      validColumns.add(
+          convertColumnToTabletRepresentation(
+              dataTypes[i],
+              columns[i],
+              Objects.nonNull(bitMaps) && i < bitMaps.length ? bitMaps[i] : null));
       if (Objects.nonNull(validBitMaps)) {
         validBitMaps.add(i < bitMaps.length ? bitMaps[i] : null);
       }
@@ -709,9 +719,14 @@ private void addValueToTablet(
         ((boolean[]) tablet.getValues()[columnIndex])[rowIndex] = (boolean) value;
         break;
       case INT32:
-      case DATE:
         ((int[]) tablet.getValues()[columnIndex])[rowIndex] = (int) value;
         break;
+      case DATE:
+        ((LocalDate[]) tablet.getValues()[columnIndex])[rowIndex] =
+            value instanceof LocalDate
+                ? (LocalDate) value
+                : DateUtils.parseIntToLocalDate((int) value);
+        break;
       case INT64:
       case TIMESTAMP:
         ((long[]) tablet.getValues()[columnIndex])[rowIndex] = (long) value;
@@ -753,10 +768,15 @@ private void copyColumnValue(
             ((boolean[]) sourceColumn)[sourceRowIndex];
         break;
       case INT32:
-      case DATE:
         ((int[]) tablet.getValues()[targetColumnIndex])[targetRowIndex] =
             ((int[]) sourceColumn)[sourceRowIndex];
         break;
+      case DATE:
+        ((LocalDate[]) tablet.getValues()[targetColumnIndex])[targetRowIndex] =
+            sourceColumn instanceof LocalDate[]
+                ? ((LocalDate[]) sourceColumn)[sourceRowIndex]
+                : DateUtils.parseIntToLocalDate(((int[]) sourceColumn)[sourceRowIndex]);
+        break;
       case INT64:
       case TIMESTAMP:
         ((long[]) tablet.getValues()[targetColumnIndex])[targetRowIndex] =
@@ -788,6 +808,22 @@ private void copyColumnValue(
     }
   }
 
+  private Object convertColumnToTabletRepresentation(
+      final TSDataType dataType, final Object column, final BitMap bitMap) {
+    if (dataType != TSDataType.DATE || !(column instanceof int[])) {
+      return column;
+    }
+
+    final int[] dateValues = (int[]) column;
+    final LocalDate[] localDateValues = new LocalDate[dateValues.length];
+    for (int i = 0; i < dateValues.length; i++) {
+      if (Objects.isNull(bitMap) || !bitMap.isMarked(i)) {
+        localDateValues[i] = DateUtils.parseIntToLocalDate(dateValues[i]);
+      }
+    }
+    return localDateValues;
+  }
+
   private static final class MatchedTreeRow {
 
     private final InsertRowNode rowNode;

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java

@@ -24,6 +24,7 @@
 import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
 import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode;
@@ -37,12 +38,14 @@
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.utils.Binary;
 import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.utils.DateUtils;
 import org.apache.tsfile.write.record.Tablet;
 import org.apache.tsfile.write.schema.MeasurementSchema;
 import org.junit.Assert;
 import org.junit.Test;
 
 import java.nio.charset.StandardCharsets;
+import java.time.LocalDate;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
@@ -110,6 +113,36 @@ public void testConvertRelationalInsertTabletNodeWithSingleMatchedColumn() {
     Assert.assertSame(node.getColumns()[2], tablet.getValues()[1]);
   }
 
+  @Test
+  public void testConvertRelationalInsertTabletNodeConvertsDateForSerialization() throws Exception {
+    final ConsensusLogToTabletConverter converter = createConverter("value");
+    final LocalDate date = LocalDate.of(2026, 1, 31);
+    final RelationalInsertTabletNode node =
+        new RelationalInsertTabletNode(
+            new PlanNodeId("date"),
+            new PartialPath(new String[] {StatementTestUtils.tableName()}),
+            true,
+            new String[] {"device_id", "value"},
+            new TSDataType[] {TSDataType.STRING, TSDataType.DATE},
+            new long[] {1L},
+            null,
+            new Object[] {
+              new Binary[] {new Binary("device_0".getBytes(StandardCharsets.UTF_8))},
+              new int[] {DateUtils.parseDateExpressionToInt(date)}
+            },
+            1,
+            new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD});
+
+    final List<Tablet> tablets = converter.convert(node);
+
+    Assert.assertEquals(1, tablets.size());
+    final Tablet tablet = tablets.get(0);
+    Assert.assertArrayEquals(new LocalDate[] {date}, (LocalDate[]) tablet.getValues()[1]);
+    final Tablet deserializedTablet = Tablet.deserialize(tablet.serialize());
+    Assert.assertArrayEquals(
+        new LocalDate[] {date}, (LocalDate[]) deserializedTablet.getValues()[1]);
+  }
+
   @Test
   public void testConvertRelationalInsertTabletNodeKeepsNullMatchedFieldColumn() {
     final ConsensusLogToTabletConverter converter = createConverter("m1");