You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2025/02/26 03:01:17 UTC
(iotdb) branch enhance_wal_compatibility created (now 5ea116583f3)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a change to branch enhance_wal_compatibility
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 5ea116583f3 enhance wal compatibility
This branch includes the following new commits:
new 5ea116583f3 enhance wal compatibility
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: enhance wal compatibility
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch enhance_wal_compatibility
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5ea116583f32846d6949722b51280dad7a139eff
Author: HTHou <ha...@apache.org>
AuthorDate: Wed Feb 26 10:41:40 2025 +0800
enhance wal compatibility
---
.../dataregion/memtable/AbstractMemTable.java | 52 ++++++++++++++++------
.../memtable/AlignedWritableMemChunk.java | 13 ++++++
.../memtable/AlignedWritableMemChunkGroup.java | 10 ++++-
.../dataregion/memtable/WritableMemChunk.java | 8 ++++
.../dataregion/memtable/WritableMemChunkGroup.java | 12 +++++
.../iotdb/db/utils/datastructure/BinaryTVList.java | 13 ++++++
.../db/utils/datastructure/BooleanTVList.java | 13 ++++++
.../iotdb/db/utils/datastructure/DoubleTVList.java | 13 ++++++
.../iotdb/db/utils/datastructure/FloatTVList.java | 13 ++++++
.../iotdb/db/utils/datastructure/IntTVList.java | 13 ++++++
.../iotdb/db/utils/datastructure/LongTVList.java | 13 ++++++
.../iotdb/db/utils/datastructure/TVList.java | 27 ++++++++++-
12 files changed, 184 insertions(+), 16 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
index 9974eee02a7..ab6fba9b756 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
@@ -843,10 +843,12 @@ public abstract class AbstractMemTable implements IMemTable {
/** Notice: this method is concurrent unsafe. */
@Override
public void serializeToWAL(IWALByteBufferView buffer) {
- WALWriteUtils.write(isSignalMemTable(), buffer);
if (isSignalMemTable()) {
+ WALWriteUtils.write(isSignalMemTable(), buffer);
return;
}
+ // marker for multi-tvlist mem chunk
+ WALWriteUtils.write((byte) -1, buffer);
buffer.putInt(seriesNumber);
buffer.putLong(memSize);
buffer.putLong(tvListRamCost);
@@ -868,7 +870,8 @@ public abstract class AbstractMemTable implements IMemTable {
}
}
- public void deserialize(DataInputStream stream) throws IOException {
+ protected void deserialize(DataInputStream stream, boolean multiTvListInMemChunk)
+ throws IOException {
seriesNumber = stream.readInt();
memSize = stream.readLong();
tvListRamCost = stream.readLong();
@@ -882,16 +885,27 @@ public abstract class AbstractMemTable implements IMemTable {
IDeviceID deviceID = Deserializer.DEFAULT_DESERIALIZER.deserializeFrom(stream);
boolean isAligned = ReadWriteIOUtils.readBool(stream);
IWritableMemChunkGroup memChunkGroup;
- if (isAligned) {
- memChunkGroup = AlignedWritableMemChunkGroup.deserialize(stream, deviceID.isTableModel());
+ if (multiTvListInMemChunk) {
+ if (isAligned) {
+ memChunkGroup = AlignedWritableMemChunkGroup.deserialize(stream, deviceID.isTableModel());
+ } else {
+ memChunkGroup = WritableMemChunkGroup.deserialize(stream);
+ }
} else {
- memChunkGroup = WritableMemChunkGroup.deserialize(stream);
+ if (isAligned) {
+ memChunkGroup =
+ AlignedWritableMemChunkGroup.deserializeSingleTVListMemChunks(
+ stream, deviceID.isTableModel());
+ } else {
+ memChunkGroup = WritableMemChunkGroup.deserializeSingleTVListMemChunks(stream);
+ }
}
memTableMap.put(deviceID, memChunkGroup);
}
}
- public void deserializeFromOldMemTableSnapshot(DataInputStream stream) throws IOException {
+ public void deserializeFromOldMemTableSnapshot(
+ DataInputStream stream, boolean multiTvListInMemChunk) throws IOException {
seriesNumber = stream.readInt();
memSize = stream.readLong();
tvListRamCost = stream.readLong();
@@ -913,10 +927,20 @@ public abstract class AbstractMemTable implements IMemTable {
IDeviceID deviceID = deviceIDFactory.getDeviceID(devicePath);
boolean isAligned = ReadWriteIOUtils.readBool(stream);
IWritableMemChunkGroup memChunkGroup;
- if (isAligned) {
- memChunkGroup = AlignedWritableMemChunkGroup.deserialize(stream, false);
+ if (multiTvListInMemChunk) {
+ if (isAligned) {
+ memChunkGroup = AlignedWritableMemChunkGroup.deserialize(stream, deviceID.isTableModel());
+ } else {
+ memChunkGroup = WritableMemChunkGroup.deserialize(stream);
+ }
} else {
- memChunkGroup = WritableMemChunkGroup.deserialize(stream);
+ if (isAligned) {
+ memChunkGroup =
+ AlignedWritableMemChunkGroup.deserializeSingleTVListMemChunks(
+ stream, deviceID.isTableModel());
+ } else {
+ memChunkGroup = WritableMemChunkGroup.deserializeSingleTVListMemChunks(stream);
+ }
}
memTableMap.put(deviceID, memChunkGroup);
}
@@ -941,14 +965,15 @@ public abstract class AbstractMemTable implements IMemTable {
}
public static IMemTable create(DataInputStream stream) throws IOException {
- boolean isSignal = ReadWriteIOUtils.readBool(stream);
+ byte marker = ReadWriteIOUtils.readByte(stream);
+ boolean isSignal = marker == 1;
IMemTable memTable;
if (isSignal) {
memTable = new NotifyFlushMemTable();
} else {
// database will be updated when deserialize
PrimitiveMemTable primitiveMemTable = new PrimitiveMemTable();
- primitiveMemTable.deserialize(stream);
+ primitiveMemTable.deserialize(stream, marker == -1);
memTable = primitiveMemTable;
}
return memTable;
@@ -956,14 +981,15 @@ public abstract class AbstractMemTable implements IMemTable {
public static IMemTable createFromOldMemTableSnapshot(DataInputStream stream)
throws IOException {
- boolean isSignal = ReadWriteIOUtils.readBool(stream);
+ byte marker = ReadWriteIOUtils.readByte(stream);
+ boolean isSignal = marker == 1;
IMemTable memTable;
if (isSignal) {
memTable = new NotifyFlushMemTable();
} else {
// database will be updated when deserialize
PrimitiveMemTable primitiveMemTable = new PrimitiveMemTable();
- primitiveMemTable.deserializeFromOldMemTableSnapshot(stream);
+ primitiveMemTable.deserializeFromOldMemTableSnapshot(stream, marker == -1);
memTable = primitiveMemTable;
}
return memTable;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
index 4d96e9cca3f..cb68440514a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
@@ -988,6 +988,19 @@ public class AlignedWritableMemChunk implements IWritableMemChunk {
return chunk;
}
+ public static AlignedWritableMemChunk deserializeSingleTVListMemChunks(
+ DataInputStream stream, boolean isTableModel) throws IOException {
+ int schemaListSize = stream.readInt();
+ List<IMeasurementSchema> schemaList = new ArrayList<>(schemaListSize);
+ for (int i = 0; i < schemaListSize; i++) {
+ IMeasurementSchema schema = MeasurementSchema.deserializeFrom(stream);
+ schemaList.add(schema);
+ }
+
+ AlignedTVList list = AlignedTVList.deserialize(stream);
+ return new AlignedWritableMemChunk(schemaList, list, isTableModel);
+ }
+
public List<IMeasurementSchema> getSchemaList() {
return schemaList;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
index 8f22d33d196..cd90daf4870 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
@@ -161,10 +161,18 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup {
memChunk.serializeToWAL(buffer);
}
- public static AlignedWritableMemChunkGroup deserialize(
+ protected static AlignedWritableMemChunkGroup deserialize(
DataInputStream stream, boolean isTableModel) throws IOException {
AlignedWritableMemChunkGroup memChunkGroup = new AlignedWritableMemChunkGroup();
memChunkGroup.memChunk = AlignedWritableMemChunk.deserialize(stream, isTableModel);
return memChunkGroup;
}
+
+ protected static AlignedWritableMemChunkGroup deserializeSingleTVListMemChunks(
+ DataInputStream stream, boolean isTableModel) throws IOException {
+ AlignedWritableMemChunkGroup memChunkGroup = new AlignedWritableMemChunkGroup();
+ memChunkGroup.memChunk =
+ AlignedWritableMemChunk.deserializeSingleTVListMemChunks(stream, isTableModel);
+ return memChunkGroup;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
index ed69cdaa681..be42352dfea 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
@@ -760,6 +760,14 @@ public class WritableMemChunk implements IWritableMemChunk {
return memChunk;
}
+ public static WritableMemChunk deserializeSingleTVListMemChunks(DataInputStream stream)
+ throws IOException {
+ WritableMemChunk memChunk = new WritableMemChunk();
+ memChunk.schema = MeasurementSchema.deserializeFrom(stream);
+ memChunk.list = TVList.deserialize(stream);
+ return memChunk;
+ }
+
@Override
public List<TVList> getSortedList() {
return sortedList;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
index cd2be43025e..0da39435d32 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
@@ -193,4 +193,16 @@ public class WritableMemChunkGroup implements IWritableMemChunkGroup {
}
return memChunkGroup;
}
+
+ public static WritableMemChunkGroup deserializeSingleTVListMemChunks(DataInputStream stream)
+ throws IOException {
+ WritableMemChunkGroup memChunkGroup = new WritableMemChunkGroup();
+ int memChunkMapSize = stream.readInt();
+ for (int i = 0; i < memChunkMapSize; ++i) {
+ String measurement = ReadWriteIOUtils.readString(stream);
+ IWritableMemChunk memChunk = WritableMemChunk.deserializeSingleTVListMemChunks(stream);
+ memChunkGroup.memChunkMap.put(measurement, memChunk);
+ }
+ return memChunkGroup;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
index cf1808b5a73..5e1510e6467 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
@@ -308,4 +308,17 @@ public abstract class BinaryTVList extends TVList {
tvList.putBinaries(times, values, bitMap, 0, rowCount);
return tvList;
}
+
+ public static BinaryTVList deserializeWithoutBitMap(DataInputStream stream) throws IOException {
+ BinaryTVList tvList = BinaryTVList.newList();
+ int rowCount = stream.readInt();
+ long[] times = new long[rowCount];
+ Binary[] values = new Binary[rowCount];
+ for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) {
+ times[rowIdx] = stream.readLong();
+ values[rowIdx] = ReadWriteIOUtils.readBinary(stream);
+ }
+ tvList.putBinaries(times, values, null, 0, rowCount);
+ return tvList;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
index 0285c37bef7..9da6cde836e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
@@ -299,4 +299,17 @@ public abstract class BooleanTVList extends TVList {
tvList.putBooleans(times, values, bitMap, 0, rowCount);
return tvList;
}
+
+ public static BooleanTVList deserializeWithoutBitMap(DataInputStream stream) throws IOException {
+ BooleanTVList tvList = BooleanTVList.newList();
+ int rowCount = stream.readInt();
+ long[] times = new long[rowCount];
+ boolean[] values = new boolean[rowCount];
+ for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) {
+ times[rowIdx] = stream.readLong();
+ values[rowIdx] = ReadWriteIOUtils.readBool(stream);
+ }
+ tvList.putBooleans(times, values, null, 0, rowCount);
+ return tvList;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
index a8bd715fee9..d94e234d162 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
@@ -305,4 +305,17 @@ public abstract class DoubleTVList extends TVList {
tvList.putDoubles(times, values, bitMap, 0, rowCount);
return tvList;
}
+
+ public static DoubleTVList deserializeWithoutBitMap(DataInputStream stream) throws IOException {
+ DoubleTVList tvList = DoubleTVList.newList();
+ int rowCount = stream.readInt();
+ long[] times = new long[rowCount];
+ double[] values = new double[rowCount];
+ for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) {
+ times[rowIdx] = stream.readLong();
+ values[rowIdx] = stream.readDouble();
+ }
+ tvList.putDoubles(times, values, null, 0, rowCount);
+ return tvList;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
index 1cf253f19ac..ba4989190d6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
@@ -305,4 +305,17 @@ public abstract class FloatTVList extends TVList {
tvList.putFloats(times, values, bitMap, 0, rowCount);
return tvList;
}
+
+ public static FloatTVList deserializeWithoutBitMap(DataInputStream stream) throws IOException {
+ FloatTVList tvList = FloatTVList.newList();
+ int rowCount = stream.readInt();
+ long[] times = new long[rowCount];
+ float[] values = new float[rowCount];
+ for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) {
+ times[rowIdx] = stream.readLong();
+ values[rowIdx] = stream.readFloat();
+ }
+ tvList.putFloats(times, values, null, 0, rowCount);
+ return tvList;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
index 72a8e1d83ff..078c85f9ce8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
@@ -297,4 +297,17 @@ public abstract class IntTVList extends TVList {
tvList.putInts(times, values, bitMap, 0, rowCount);
return tvList;
}
+
+ public static IntTVList deserializeWithoutBitMap(DataInputStream stream) throws IOException {
+ IntTVList tvList = IntTVList.newList();
+ int rowCount = stream.readInt();
+ long[] times = new long[rowCount];
+ int[] values = new int[rowCount];
+ for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) {
+ times[rowIdx] = stream.readLong();
+ values[rowIdx] = stream.readInt();
+ }
+ tvList.putInts(times, values, null, 0, rowCount);
+ return tvList;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
index 7dc063a2db5..ed9671c7f11 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
@@ -297,4 +297,17 @@ public abstract class LongTVList extends TVList {
tvList.putLongs(times, values, bitMap, 0, rowCount);
return tvList;
}
+
+ public static LongTVList deserializeWithoutBitMap(DataInputStream stream) throws IOException {
+ LongTVList tvList = LongTVList.newList();
+ int rowCount = stream.readInt();
+ long[] times = new long[rowCount];
+ long[] values = new long[rowCount];
+ for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) {
+ times[rowIdx] = stream.readLong();
+ values[rowIdx] = stream.readLong();
+ }
+ tvList.putLongs(times, values, null, 0, rowCount);
+ return tvList;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
index a04580f429a..4e5d70519b7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
@@ -560,8 +560,31 @@ public abstract class TVList implements WALEntryValue {
return DoubleTVList.deserialize(stream);
case BOOLEAN:
return BooleanTVList.deserialize(stream);
- case VECTOR:
- return AlignedTVList.deserialize(stream);
+ default:
+ break;
+ }
+ return null;
+ }
+
+ public static TVList deserializeWithoutBitMap(DataInputStream stream) throws IOException {
+ TSDataType dataType = ReadWriteIOUtils.readDataType(stream);
+ switch (dataType) {
+ case TEXT:
+ case BLOB:
+ case STRING:
+ return BinaryTVList.deserializeWithoutBitMap(stream);
+ case FLOAT:
+ return FloatTVList.deserializeWithoutBitMap(stream);
+ case INT32:
+ case DATE:
+ return IntTVList.deserializeWithoutBitMap(stream);
+ case INT64:
+ case TIMESTAMP:
+ return LongTVList.deserializeWithoutBitMap(stream);
+ case DOUBLE:
+ return DoubleTVList.deserializeWithoutBitMap(stream);
+ case BOOLEAN:
+ return BooleanTVList.deserializeWithoutBitMap(stream);
default:
break;
}