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 c32431faa5d Use TypeService for ImportWAL snapshot replay (#18711)
c32431faa5d is described below
commit c32431faa5d78d04998a2a04825c9f5b706ac925
Author: Caideyipi <[email protected]>
AuthorDate: Thu Sep 24 11:58:32 2026 +0800
Use TypeService for ImportWAL snapshot replay (#18711)
---
.../java/org/apache/iotdb/db/tools/ImportWAL.java | 62 +++++------------
.../org/apache/iotdb/db/utils/TypeServices.java | 81 ++++++++++++++++++++++
.../org/apache/iotdb/db/tools/ImportWALTest.java | 46 ++++++++++--
3 files changed, 139 insertions(+), 50 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java
index d3b00bf87c0..265c027e8ac 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java
@@ -43,6 +43,7 @@ import
org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
import
org.apache.iotdb.db.subscription.broker.consensus.ConsensusLogToTabletConverter;
import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
import org.apache.iotdb.db.tools.TableWALDeleteConverter.ConversionException;
+import org.apache.iotdb.db.utils.TypeServices;
import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.iotdb.isession.SessionDataSet;
@@ -60,9 +61,9 @@ import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.type.Type;
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;
@@ -74,7 +75,6 @@ import java.io.PrintWriter;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
-import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
@@ -1599,16 +1599,10 @@ public class ImportWAL {
}
private static Object createValueArray(final TSDataType type, final int
rowCount) {
- return switch (type) {
- case BOOLEAN -> new boolean[rowCount];
- case INT32 -> new int[rowCount];
- case DATE -> new LocalDate[rowCount];
- case INT64, TIMESTAMP -> new long[rowCount];
- case FLOAT -> new float[rowCount];
- case DOUBLE -> new double[rowCount];
- case TEXT, STRING, BLOB, OBJECT -> new Binary[rowCount];
- case VECTOR, UNKNOWN -> throw unsupportedSnapshotDataType(type);
- };
+ ensureSupportedSnapshotDataType(type);
+ return TypeServices.StorageEngine.TABLET_COLUMN_ALLOCATOR_SERVICE
+ .call(Type.fromTsDataType(type))
+ .apply(rowCount);
}
private static Object[] createValueArrays(
@@ -1626,19 +1620,10 @@ public class ImportWAL {
final TSDataType type,
final TVList list,
final int sourceIndex) {
- switch (type) {
- case BOOLEAN -> ((boolean[]) target)[targetIndex] =
list.getBoolean(sourceIndex);
- case INT32 -> ((int[]) target)[targetIndex] = list.getInt(sourceIndex);
- case DATE ->
- ((LocalDate[]) target)[targetIndex] =
- DateUtils.parseIntToLocalDate(list.getInt(sourceIndex));
- case INT64, TIMESTAMP -> ((long[]) target)[targetIndex] =
list.getLong(sourceIndex);
- case FLOAT -> ((float[]) target)[targetIndex] =
list.getFloat(sourceIndex);
- case DOUBLE -> ((double[]) target)[targetIndex] =
list.getDouble(sourceIndex);
- case TEXT, STRING, BLOB, OBJECT ->
- ((Binary[]) target)[targetIndex] = list.getBinary(sourceIndex);
- case VECTOR, UNKNOWN -> throw unsupportedSnapshotDataType(type);
- }
+ ensureSupportedSnapshotDataType(type);
+ TypeServices.StorageEngine.TV_LIST_TABLET_VALUE_WRITER_SERVICE
+ .call(Type.fromTsDataType(type))
+ .write(target, targetIndex, list, sourceIndex);
}
private static void putValue(
@@ -1648,24 +1633,15 @@ public class ImportWAL {
final AlignedTVList list,
final int sourceIndex,
final int columnIndex) {
- switch (type) {
- case BOOLEAN ->
- ((boolean[]) target)[targetIndex] =
- list.getBooleanByValueIndex(sourceIndex, columnIndex);
- case INT32 ->
- ((int[]) target)[targetIndex] =
list.getIntByValueIndex(sourceIndex, columnIndex);
- case DATE ->
- ((LocalDate[]) target)[targetIndex] =
-
DateUtils.parseIntToLocalDate(list.getIntByValueIndex(sourceIndex,
columnIndex));
- case INT64, TIMESTAMP ->
- ((long[]) target)[targetIndex] =
list.getLongByValueIndex(sourceIndex, columnIndex);
- case FLOAT ->
- ((float[]) target)[targetIndex] =
list.getFloatByValueIndex(sourceIndex, columnIndex);
- case DOUBLE ->
- ((double[]) target)[targetIndex] =
list.getDoubleByValueIndex(sourceIndex, columnIndex);
- case TEXT, STRING, BLOB, OBJECT ->
- ((Binary[]) target)[targetIndex] =
list.getBinaryByValueIndex(sourceIndex, columnIndex);
- case VECTOR, UNKNOWN -> throw unsupportedSnapshotDataType(type);
+ ensureSupportedSnapshotDataType(type);
+ TypeServices.StorageEngine.ALIGNED_TV_LIST_TABLET_VALUE_WRITER_SERVICE
+ .call(Type.fromTsDataType(type))
+ .write(target, targetIndex, list, sourceIndex, columnIndex);
+ }
+
+ private static void ensureSupportedSnapshotDataType(final TSDataType type)
{
+ if (type == TSDataType.VECTOR || type == TSDataType.UNKNOWN) {
+ throw unsupportedSnapshotDataType(type);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeServices.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeServices.java
index a4431772b31..15634f54db1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeServices.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeServices.java
@@ -2884,6 +2884,74 @@ public class TypeServices {
.setChecked(true);
};
+ public static final TypeService<TVListTabletValueWriter>
TV_LIST_TABLET_VALUE_WRITER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN ->
+ (target, targetIndex, list, sourceIndex) ->
+ ((boolean[]) target)[targetIndex] =
list.getBoolean(sourceIndex);
+ case INT32 ->
+ (target, targetIndex, list, sourceIndex) ->
+ ((int[]) target)[targetIndex] = list.getInt(sourceIndex);
+ case DATE ->
+ (target, targetIndex, list, sourceIndex) ->
+ ((LocalDate[]) target)[targetIndex] =
+
DateUtils.parseIntToLocalDate(list.getInt(sourceIndex));
+ case INT64, TIMESTAMP ->
+ (target, targetIndex, list, sourceIndex) ->
+ ((long[]) target)[targetIndex] =
list.getLong(sourceIndex);
+ case FLOAT ->
+ (target, targetIndex, list, sourceIndex) ->
+ ((float[]) target)[targetIndex] =
list.getFloat(sourceIndex);
+ case DOUBLE ->
+ (target, targetIndex, list, sourceIndex) ->
+ ((double[]) target)[targetIndex] =
list.getDouble(sourceIndex);
+ case TEXT, BLOB, STRING, OBJECT ->
+ (target, targetIndex, list, sourceIndex) ->
+ ((Binary[]) target)[targetIndex] =
list.getBinary(sourceIndex);
+ case ROW, UNKNOWN, VECTOR ->
+ throw new
UnSupportedDataTypeException(type.getTypeEnum().name())
+ .setChecked(true);
+ };
+
+ public static final TypeService<AlignedTVListTabletValueWriter>
+ ALIGNED_TV_LIST_TABLET_VALUE_WRITER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN ->
+ (target, targetIndex, list, sourceIndex, columnIndex) ->
+ ((boolean[]) target)[targetIndex] =
+ list.getBooleanByValueIndex(sourceIndex,
columnIndex);
+ case INT32 ->
+ (target, targetIndex, list, sourceIndex, columnIndex) ->
+ ((int[]) target)[targetIndex] =
+ list.getIntByValueIndex(sourceIndex,
columnIndex);
+ case DATE ->
+ (target, targetIndex, list, sourceIndex, columnIndex) ->
+ ((LocalDate[]) target)[targetIndex] =
+ DateUtils.parseIntToLocalDate(
+ list.getIntByValueIndex(sourceIndex,
columnIndex));
+ case INT64, TIMESTAMP ->
+ (target, targetIndex, list, sourceIndex, columnIndex) ->
+ ((long[]) target)[targetIndex] =
+ list.getLongByValueIndex(sourceIndex,
columnIndex);
+ case FLOAT ->
+ (target, targetIndex, list, sourceIndex, columnIndex) ->
+ ((float[]) target)[targetIndex] =
+ list.getFloatByValueIndex(sourceIndex,
columnIndex);
+ case DOUBLE ->
+ (target, targetIndex, list, sourceIndex, columnIndex) ->
+ ((double[]) target)[targetIndex] =
+ list.getDoubleByValueIndex(sourceIndex,
columnIndex);
+ case TEXT, BLOB, STRING, OBJECT ->
+ (target, targetIndex, list, sourceIndex, columnIndex) ->
+ ((Binary[]) target)[targetIndex] =
+ list.getBinaryByValueIndex(sourceIndex,
columnIndex);
+ case ROW, UNKNOWN, VECTOR ->
+ throw new
UnSupportedDataTypeException(type.getTypeEnum().name())
+ .setChecked(true);
+ };
+
public static final TypeService<IntFunction<Object>>
EMPTY_TABLET_COLUMN_FACTORY_SERVICE =
type ->
switch (type.getTypeEnum()) {
@@ -3121,6 +3189,8 @@ public class TypeServices {
PRIMITIVE_ARRAY_ALLOCATOR_SERVICE.check();
SOURCE_COLUMN_TO_TABLET_VALUE_WRITER_SERVICE.check();
TABLET_COLUMN_ALLOCATOR_SERVICE.check();
+ TV_LIST_TABLET_VALUE_WRITER_SERVICE.check();
+ ALIGNED_TV_LIST_TABLET_VALUE_WRITER_SERVICE.check();
EMPTY_TABLET_COLUMN_FACTORY_SERVICE.check();
WINDOW_VALUE_ARRAY_BUILDER_SERVICE.check();
RAW_ARRAY_BYTE_BUFFER_DESERIALIZER_SERVICE.check();
@@ -4311,6 +4381,17 @@ public class TypeServices {
int targetIndex);
}
+ @FunctionalInterface
+ public interface TVListTabletValueWriter {
+ void write(Object target, int targetIndex, TVList list, int sourceIndex);
+ }
+
+ @FunctionalInterface
+ public interface AlignedTVListTabletValueWriter {
+ void write(
+ Object target, int targetIndex, AlignedTVList list, int sourceIndex,
int columnIndex);
+ }
+
@FunctionalInterface
public interface DecodedValueChunkWriter {
void write(ValueChunkWriter writer, long time, InputStream stream, boolean
isNull)
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java
index 0cce7ee2bd1..aa71595d443 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java
@@ -58,6 +58,7 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.StringArrayDeviceID;
import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.utils.Binary;
+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;
@@ -75,6 +76,7 @@ import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -1528,16 +1530,27 @@ public class ImportWALTest {
final List<IMeasurementSchema> schemas =
Arrays.asList(
new MeasurementSchema("s1", TSDataType.INT32),
- new MeasurementSchema("s2", TSDataType.INT64));
+ new MeasurementSchema("s2", TSDataType.INT64),
+ new MeasurementSchema("s3", TSDataType.DATE));
+ final LocalDate firstDate = LocalDate.of(2026, 9, 23);
+ final LocalDate secondDate = LocalDate.of(2026, 9, 24);
final StringArrayDeviceID deviceId = new StringArrayDeviceID("root.sg.d1");
- memTable.write(deviceId, schemas, 3, new Object[] {30, 300L});
- memTable.write(deviceId, schemas, 1, new Object[] {10, null});
+ memTable.write(
+ deviceId,
+ schemas,
+ 3,
+ new Object[] {30, 300L,
DateUtils.parseDateExpressionToInt(secondDate)});
+ memTable.write(
+ deviceId,
+ schemas,
+ 1,
+ new Object[] {10, null,
DateUtils.parseDateExpressionToInt(firstDate)});
final Session treeSession = mock(Session.class);
new ImportWAL.WALReplayer(treeSession, null, null).replay(new
WALInfoEntry(1, memTable));
final ArgumentCaptor<Tablet> tabletCaptor =
ArgumentCaptor.forClass(Tablet.class);
- verify(treeSession, times(2)).insertTablet(tabletCaptor.capture());
+ verify(treeSession, times(3)).insertTablet(tabletCaptor.capture());
final Tablet s1Tablet =
tabletCaptor.getAllValues().stream()
.filter(tablet ->
"s1".equals(tablet.getSchemas().get(0).getMeasurementName()))
@@ -1547,6 +1560,13 @@ public class ImportWALTest {
assertEquals(1, s1Tablet.getTimestamp(0));
assertEquals(3, s1Tablet.getTimestamp(1));
assertArrayEquals(new int[] {10, 30}, (int[]) s1Tablet.getValues()[0]);
+ final Tablet s3Tablet =
+ tabletCaptor.getAllValues().stream()
+ .filter(tablet ->
"s3".equals(tablet.getSchemas().get(0).getMeasurementName()))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ assertArrayEquals(
+ new LocalDate[] {firstDate, secondDate}, (LocalDate[])
s3Tablet.getValues()[0]);
}
/** Covers an aligned snapshot with nulls and verifies the aligned Session
API is used. */
@@ -1556,10 +1576,21 @@ public class ImportWALTest {
final List<IMeasurementSchema> schemas =
Arrays.asList(
new MeasurementSchema("s1", TSDataType.INT32),
- new MeasurementSchema("s2", TSDataType.INT64));
+ new MeasurementSchema("s2", TSDataType.INT64),
+ new MeasurementSchema("s3", TSDataType.DATE));
+ final LocalDate firstDate = LocalDate.of(2026, 9, 23);
+ final LocalDate secondDate = LocalDate.of(2026, 9, 24);
final StringArrayDeviceID deviceId = new StringArrayDeviceID("root.sg.d1");
- memTable.writeAlignedRow(deviceId, schemas, 2, new Object[] {20, null});
- memTable.writeAlignedRow(deviceId, schemas, 1, new Object[] {10, 100L});
+ memTable.writeAlignedRow(
+ deviceId,
+ schemas,
+ 2,
+ new Object[] {20, null,
DateUtils.parseDateExpressionToInt(secondDate)});
+ memTable.writeAlignedRow(
+ deviceId,
+ schemas,
+ 1,
+ new Object[] {10, 100L,
DateUtils.parseDateExpressionToInt(firstDate)});
final Session treeSession = mock(Session.class);
new ImportWAL.WALReplayer(treeSession, null, null).replay(new
WALInfoEntry(1, memTable));
@@ -1573,6 +1604,7 @@ public class ImportWALTest {
assertEquals(2, tablet.getTimestamp(1));
assertArrayEquals(new int[] {10, 20}, (int[]) tablet.getValues()[0]);
assertTrue(tablet.getBitMaps()[1].isMarked(1));
+ assertArrayEquals(new LocalDate[] {firstDate, secondDate}, (LocalDate[])
tablet.getValues()[2]);
}
/** Covers snapshot serialization and deserialization through a real WAL
file. */
