You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2024/10/29 03:59:42 UTC
(iotdb) branch master updated: No need to split relationalInsertTabletNode when serialize to WAL (#13929)
This is an automated email from the ASF dual-hosted git repository.
jiangtian 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 5726de01b9f No need to split relationalInsertTabletNode when serialize to WAL (#13929)
5726de01b9f is described below
commit 5726de01b9fbd74bc1d25bd782a3fdf12372e626
Author: Haonan <hh...@outlook.com>
AuthorDate: Tue Oct 29 11:59:36 2024 +0800
No need to split relationalInsertTabletNode when serialize to WAL (#13929)
* No need to split insertTabletNode when serialize to WAL
* add UT
---
.../planner/plan/node/write/InsertTabletNode.java | 46 ++++++++++------
.../node/write/RelationalInsertTabletNode.java | 4 +-
.../dataregion/wal/buffer/WALInfoEntry.java | 10 +---
.../dataregion/wal/io/WALFileTest.java | 64 ++++++++++++++++++++++
4 files changed, 98 insertions(+), 26 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java
index 98a072fcee0..093f3715957 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java
@@ -838,22 +838,26 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
*/
@Override
public void serializeToWAL(IWALByteBufferView buffer) {
- serializeToWAL(buffer, 0, rowCount);
+ serializeToWAL(buffer, Collections.singletonList(new int[] {0, rowCount}));
}
- public void serializeToWAL(IWALByteBufferView buffer, int start, int end) {
+ public void serializeToWAL(IWALByteBufferView buffer, List<int[]> rangeList) {
buffer.putShort(getType().getNodeType());
- subSerialize(buffer, start, end);
+ subSerialize(buffer, rangeList);
}
- void subSerialize(IWALByteBufferView buffer, int start, int end) {
+ void subSerialize(IWALByteBufferView buffer, List<int[]> rangeList) {
buffer.putLong(searchIndex);
WALWriteUtils.write(targetPath.getFullPath(), buffer);
// data types are serialized in measurement schemas
writeMeasurementSchemas(buffer);
- writeTimes(buffer, start, end);
- writeBitMaps(buffer, start, end);
- writeValues(buffer, start, end);
+ int rowNumInRange = 0;
+ for (int[] startEnd : rangeList) {
+ rowNumInRange += startEnd[1] - startEnd[0];
+ }
+ writeTimes(buffer, rangeList, rowNumInRange);
+ writeBitMaps(buffer, rangeList, rowNumInRange);
+ writeValues(buffer, rangeList);
buffer.put((byte) (isAligned ? 1 : 0));
}
@@ -863,15 +867,17 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
serializeMeasurementSchemasToWAL(buffer);
}
- protected void writeTimes(IWALByteBufferView buffer, int start, int end) {
- buffer.putInt(end - start);
- for (int i = start; i < end; i++) {
- buffer.putLong(times[i]);
+ protected void writeTimes(IWALByteBufferView buffer, List<int[]> rangeList, int rowNumInRange) {
+ buffer.putInt(rowNumInRange);
+ for (int[] startEnd : rangeList) {
+ for (int i = startEnd[0]; i < startEnd[1]; i++) {
+ buffer.putLong(times[i]);
+ }
}
}
/** Serialize bitmaps, ignoring failed time series */
- protected void writeBitMaps(IWALByteBufferView buffer, int start, int end) {
+ protected void writeBitMaps(IWALByteBufferView buffer, List<int[]> rangeList, int rowNumInRange) {
buffer.put(BytesUtils.boolToByte(bitMaps != null));
if (bitMaps != null) {
for (int i = 0; i < bitMaps.length; i++) {
@@ -884,9 +890,13 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
buffer.put(BytesUtils.boolToByte(false));
} else {
buffer.put(BytesUtils.boolToByte(true));
- int len = end - start;
- BitMap partBitMap = new BitMap(len);
- BitMap.copyOfRange(bitMaps[i], start, partBitMap, 0, len);
+ BitMap partBitMap = new BitMap(rowNumInRange);
+ int copiedLength = 0;
+ for (int[] startEnd : rangeList) {
+ int len = startEnd[1] - startEnd[0];
+ BitMap.copyOfRange(bitMaps[i], startEnd[0], partBitMap, copiedLength, len);
+ copiedLength += len;
+ }
buffer.put(partBitMap.getByteArray());
}
}
@@ -894,13 +904,15 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
}
/** Serialize values, ignoring failed time series */
- protected void writeValues(IWALByteBufferView buffer, int start, int end) {
+ protected void writeValues(IWALByteBufferView buffer, List<int[]> rangeList) {
for (int i = 0; i < columns.length; i++) {
// ignore failed partial insert
if (measurements[i] == null) {
continue;
}
- serializeColumn(dataTypes[i], columns[i], buffer, start, end);
+ for (int[] startEnd : rangeList) {
+ serializeColumn(dataTypes[i], columns[i], buffer, startEnd[0], startEnd[1]);
+ }
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java
index b21cbd0a4db..35e467b7873 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java
@@ -224,8 +224,8 @@ public class RelationalInsertTabletNode extends InsertTabletNode {
}
@Override
- void subSerialize(IWALByteBufferView buffer, int start, int end) {
- super.subSerialize(buffer, start, end);
+ void subSerialize(IWALByteBufferView buffer, List<int[]> rangeList) {
+ super.subSerialize(buffer, rangeList);
for (int i = 0; i < measurements.length; i++) {
if (measurements[i] != null) {
buffer.put(columnCategories[i].getCategory());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java
index 994c0d61279..4e2ea88ce4a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java
@@ -73,21 +73,17 @@ public class WALInfoEntry extends WALEntry {
@Override
public void serialize(IWALByteBufferView buffer) {
+ buffer.put(type.getCode());
+ buffer.putLong(memTableId);
switch (type) {
case INSERT_TABLET_NODE:
- for (int[] range : tabletInfo.tabletRangeList) {
- buffer.put(type.getCode());
- buffer.putLong(memTableId);
- ((InsertTabletNode) value).serializeToWAL(buffer, range[0], range[1]);
- }
+ ((InsertTabletNode) value).serializeToWAL(buffer, tabletInfo.tabletRangeList);
break;
case INSERT_ROW_NODE:
case INSERT_ROWS_NODE:
case DELETE_DATA_NODE:
case MEMORY_TABLE_SNAPSHOT:
case CONTINUOUS_SAME_SEARCH_INDEX_SEPARATOR_NODE:
- buffer.put(type.getCode());
- buffer.putLong(memTableId);
value.serializeToWAL(buffer);
break;
case MEMORY_TABLE_CHECKPOINT:
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
index 5f05e021802..95263f0ec2c 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
@@ -21,11 +21,13 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.io;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
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.InsertTabletNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
@@ -49,6 +51,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -88,6 +91,12 @@ public class WALFileTest {
expectedWALEntries.add(new WALInfoEntry(fakeMemTableId, getInsertRowsNode(devicePath)));
expectedWALEntries.add(new WALInfoEntry(fakeMemTableId, getInsertTabletNode(devicePath)));
expectedWALEntries.add(new WALInfoEntry(fakeMemTableId, getDeleteDataNode(devicePath)));
+ expectedWALEntries.add(
+ new WALInfoEntry(
+ fakeMemTableId,
+ getRelationalInsertTabletNode("table1"),
+ Arrays.asList(new int[] {0, 2}, new int[] {2, 4})));
+
// test WALEntry.serializedSize
int size = 0;
for (WALEntry walEntry : expectedWALEntries) {
@@ -110,6 +119,14 @@ public class WALFileTest {
actualWALEntries.add(walReader.next());
}
}
+
+ // set tablet info of the RelationalInsertTablet to range [0, 4)
+ expectedWALEntries.set(
+ 4,
+ new WALInfoEntry(
+ fakeMemTableId,
+ getRelationalInsertTabletNode("table1"),
+ Collections.singletonList(new int[] {0, 4})));
assertEquals(expectedWALEntries, actualWALEntries);
}
@@ -359,6 +376,53 @@ public class WALFileTest {
times.length);
}
+ public static RelationalInsertTabletNode getRelationalInsertTabletNode(String tableName)
+ throws IllegalPathException {
+ long[] times = new long[] {110L, 111L, 112L, 113L};
+ TSDataType[] dataTypes =
+ new TSDataType[] {
+ TSDataType.STRING, TSDataType.FLOAT,
+ };
+
+ Object[] columns = new Object[2];
+ columns[0] = new Binary[4];
+ columns[1] = new float[4];
+
+ for (int r = 0; r < 4; r++) {
+ ((Binary[]) columns[0])[r] = new Binary("hh" + r, TSFileConfig.STRING_CHARSET);
+ ((float[]) columns[1])[r] = 2 + r;
+ }
+
+ BitMap[] bitMaps = new BitMap[dataTypes.length];
+ for (int i = 0; i < dataTypes.length; i++) {
+ if (bitMaps[i] == null) {
+ bitMaps[i] = new BitMap(times.length);
+ }
+ bitMaps[i].mark(i % times.length);
+ }
+ MeasurementSchema[] schemas =
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", dataTypes[0]), new MeasurementSchema("s2", dataTypes[1]),
+ };
+
+ return new RelationalInsertTabletNode(
+ new PlanNodeId(""),
+ new PartialPath(tableName, false),
+ false,
+ new String[] {
+ "s1", "s2",
+ },
+ dataTypes,
+ schemas,
+ times,
+ bitMaps,
+ columns,
+ times.length,
+ new TsTableColumnCategory[] {
+ TsTableColumnCategory.ID, TsTableColumnCategory.MEASUREMENT,
+ });
+ }
+
public static DeleteDataNode getDeleteDataNode(String devicePath) throws IllegalPathException {
DeleteDataNode deleteDataNode =
new DeleteDataNode(