From 31440d1a7ecd2426f14e6075bc78deca671f8c1c Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 16 Jul 2026 17:54:00 +0800 Subject: [PATCH 1/2] Fix consensus subscription DATE tablet conversion --- .../ConsensusLogToTabletConverter.java | 44 +++++++++++++++++-- .../ConsensusLogToTabletConverterTest.java | 33 ++++++++++++++ 2 files changed, 73 insertions(+), 4 deletions(-) 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 fd4ac07678e0..317f1f61ee93 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.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 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 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; 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 c172dbe263d0..1b8ad9683653 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.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 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"); From 18ad5977c4112f41ff39d2a7767f144941b50d36 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 17 Jul 2026 10:10:30 +0800 Subject: [PATCH 2/2] Update IoTDBConsensusSubscriptionDataTableIT.java --- ...IoTDBConsensusSubscriptionDataTableIT.java | 114 ++++++++++++++++++ 1 file changed, 114 insertions(+) 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 1be9a1eb6359..3d48bf31568b 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 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 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"); + } }