This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 5405623efe8 Subscription: Fix DATE serialization for consensus dataset
(#18227)
5405623efe8 is described below
commit 5405623efe838bfdc3616b779b39b8a0ad1a284f
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 21 10:03:36 2026 +0800
Subscription: Fix DATE serialization for consensus dataset (#18227)
* Fix consensus subscription DATE tablet conversion
* Update IoTDBConsensusSubscriptionDataTableIT.java
---
.../IoTDBConsensusSubscriptionDataTableIT.java | 114 +++++++++++++++++++++
.../consensus/ConsensusLogToTabletConverter.java | 44 +++++++-
.../ConsensusLogToTabletConverterTest.java | 33 ++++++
3 files changed, 187 insertions(+), 4 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionDataTableIT.java
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionDataTableIT.java
index 1be9a1eb635..3d48bf31568 100644
---
a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionDataTableIT.java
+++
b/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 class IoTDBConsensusSubscriptionDataTableIT extends
AbstractSubscriptionC
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");
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
index fd4ac07678e..317f1f61ee9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
@@ -48,12 +48,14 @@ import org.apache.tsfile.enums.TSDataType;
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 @@ public class ConsensusLogToTabletConverter {
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 @@ public class ConsensusLogToTabletConverter {
}
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 @@ public class ConsensusLogToTabletConverter {
((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 @@ public class ConsensusLogToTabletConverter {
((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 @@ public class ConsensusLogToTabletConverter {
}
}
+ 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;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
index c172dbe263d..1b8ad968365 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.path.PartialPath;
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.ColumnCategory;
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 class ConsensusLogToTabletConverterTest {
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");
