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 2022/09/01 13:07:57 UTC
[iotdb] branch master updated: [IOTDB-4282] Supply deserialization for DeleteDataPlan in MultiLeaderConsensus (#7205)
This is an automated email from the ASF dual-hosted git repository.
haonan 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 6d2a1cff9b [IOTDB-4282] Supply deserialization for DeleteDataPlan in MultiLeaderConsensus (#7205)
6d2a1cff9b is described below
commit 6d2a1cff9bc9965dafaa1b6f1f800debfa5ecb01
Author: Alan Choo <43...@users.noreply.github.com>
AuthorDate: Thu Sep 1 21:07:48 2022 +0800
[IOTDB-4282] Supply deserialization for DeleteDataPlan in MultiLeaderConsensus (#7205)
---
.../consensus/statemachine/BaseStateMachine.java | 2 +-
.../statemachine/visitor/DataExecutionVisitor.java | 2 +-
.../iotdb/db/engine/storagegroup/DataRegion.java | 44 +++++++++-
.../db/engine/storagegroup/TsFileProcessor.java | 5 ++
.../mpp/plan/planner/plan/node/PlanNodeType.java | 4 +
.../planner/plan/node/write/DeleteDataNode.java | 86 ++++++++++++++++++-
.../org/apache/iotdb/db/wal/buffer/WALBuffer.java | 15 ++--
.../org/apache/iotdb/db/wal/buffer/WALEntry.java | 11 ++-
.../apache/iotdb/db/wal/buffer/WALEntryType.java | 9 +-
.../apache/iotdb/db/wal/buffer/WALInfoEntry.java | 1 +
.../org/apache/iotdb/db/wal/node/IWALNode.java | 4 +
.../org/apache/iotdb/db/wal/node/WALFakeNode.java | 6 ++
.../java/org/apache/iotdb/db/wal/node/WALNode.java | 17 +++-
.../db/wal/recover/file/TsFilePlanRedoer.java | 21 +++++
.../file/UnsealedTsFileRecoverPerformer.java | 4 +
.../org/apache/iotdb/db/wal/io/WALFileTest.java | 14 +++
.../iotdb/db/wal/node/ConsensusReqReaderTest.java | 99 ++++++++++++++++------
.../db/wal/recover/file/TsFilePlanRedoerTest.java | 33 ++++++++
18 files changed, 328 insertions(+), 49 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/BaseStateMachine.java b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/BaseStateMachine.java
index 32333f26e9..759c9739e4 100644
--- a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/BaseStateMachine.java
+++ b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/BaseStateMachine.java
@@ -53,7 +53,7 @@ public abstract class BaseStateMachine implements IStateMachine, IStateMachine.E
if (request instanceof ByteBufferConsensusRequest) {
node = PlanNodeType.deserialize(request.serializeToByteBuffer());
} else if (request instanceof MultiLeaderConsensusRequest) {
- node = WALEntry.deserializeInsertNode(request.serializeToByteBuffer());
+ node = WALEntry.deserializeForConsensus(request.serializeToByteBuffer());
} else if (request instanceof PlanNode) {
node = (PlanNode) request;
} else {
diff --git a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/visitor/DataExecutionVisitor.java b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/visitor/DataExecutionVisitor.java
index 418bb2b5e2..63d5b62c2c 100644
--- a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/visitor/DataExecutionVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/visitor/DataExecutionVisitor.java
@@ -149,7 +149,7 @@ public class DataExecutionVisitor extends PlanVisitor<TSStatus, DataRegion> {
try {
for (PartialPath path : node.getPathList()) {
dataRegion.deleteByDevice(
- path, node.getDeleteStartTime(), node.getDeleteEndTime(), Long.MAX_VALUE, null);
+ path, node.getDeleteStartTime(), node.getDeleteEndTime(), node.getSearchIndex(), null);
}
return StatusUtils.OK;
} catch (IOException e) {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
index e5445125fe..e58f55202e 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
@@ -68,6 +68,8 @@ import org.apache.iotdb.db.metadata.cache.DataNodeSchemaCache;
import org.apache.iotdb.db.metadata.idtable.IDTable;
import org.apache.iotdb.db.metadata.idtable.IDTableManager;
import org.apache.iotdb.db.metadata.mnode.IMeasurementMNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertMultiTabletsNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowsNode;
@@ -2185,7 +2187,7 @@ public class DataRegion {
* @param pattern Must be a pattern start with a precise device path
* @param startTime
* @param endTime
- * @param planIndex
+ * @param searchIndex
* @param timePartitionFilter
* @throws IOException
*/
@@ -2193,7 +2195,7 @@ public class DataRegion {
PartialPath pattern,
long startTime,
long endTime,
- long planIndex,
+ long searchIndex,
TimePartitionFilter timePartitionFilter)
throws IOException {
// If there are still some old version tsfiles, the delete won't succeeded.
@@ -2225,7 +2227,7 @@ public class DataRegion {
// write log to impacted working TsFileProcessors
List<WALFlushListener> walListeners =
- logDeleteInWAL(startTime, endTime, pattern, timePartitionFilter);
+ logDeletionInWAL(startTime, endTime, searchIndex, pattern, timePartitionFilter);
for (WALFlushListener walFlushListener : walListeners) {
if (walFlushListener.waitForResult() == WALFlushListener.Status.FAILURE) {
@@ -2293,6 +2295,42 @@ public class DataRegion {
return walFlushListeners;
}
+ private List<WALFlushListener> logDeletionInWAL(
+ long startTime,
+ long endTime,
+ long searchIndex,
+ PartialPath path,
+ TimePartitionFilter timePartitionFilter) {
+ long timePartitionStartId = StorageEngine.getTimePartition(startTime);
+ long timePartitionEndId = StorageEngine.getTimePartition(endTime);
+ List<WALFlushListener> walFlushListeners = new ArrayList<>();
+ if (config.getWalMode() == WALMode.DISABLE) {
+ return walFlushListeners;
+ }
+ DeleteDataNode deleteDataNode =
+ new DeleteDataNode(new PlanNodeId(""), Collections.singletonList(path), startTime, endTime);
+ deleteDataNode.setSearchIndex(searchIndex);
+ for (Map.Entry<Long, TsFileProcessor> entry : workSequenceTsFileProcessors.entrySet()) {
+ if (timePartitionStartId <= entry.getKey()
+ && entry.getKey() <= timePartitionEndId
+ && (timePartitionFilter == null
+ || timePartitionFilter.satisfy(storageGroupName, entry.getKey()))) {
+ WALFlushListener walFlushListener = entry.getValue().logDeleteDataNodeInWAL(deleteDataNode);
+ walFlushListeners.add(walFlushListener);
+ }
+ }
+ for (Map.Entry<Long, TsFileProcessor> entry : workUnsequenceTsFileProcessors.entrySet()) {
+ if (timePartitionStartId <= entry.getKey()
+ && entry.getKey() <= timePartitionEndId
+ && (timePartitionFilter == null
+ || timePartitionFilter.satisfy(storageGroupName, entry.getKey()))) {
+ WALFlushListener walFlushListener = entry.getValue().logDeleteDataNodeInWAL(deleteDataNode);
+ walFlushListeners.add(walFlushListener);
+ }
+ }
+ return walFlushListeners;
+ }
+
private boolean canSkipDelete(
TsFileResource tsFileResource,
Set<PartialPath> devicePaths,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 45d90db709..c1f7e98f5f 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -49,6 +49,7 @@ import org.apache.iotdb.db.metadata.idtable.entry.DeviceIDFactory;
import org.apache.iotdb.db.metadata.idtable.entry.IDeviceID;
import org.apache.iotdb.db.metadata.path.AlignedPath;
import org.apache.iotdb.db.metadata.utils.ResourceByPathUtils;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
@@ -854,6 +855,10 @@ public class TsFileProcessor {
return walNode.log(workMemTable.getMemTableId(), deletePlan);
}
+ WALFlushListener logDeleteDataNodeInWAL(DeleteDataNode deleteDataNode) {
+ return walNode.log(workMemTable.getMemTableId(), deleteDataNode);
+ }
+
public TsFileResource getTsFileResource() {
return tsFileResource;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
index 349060a1bb..d7f4c3cfb9 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
@@ -161,6 +161,8 @@ public enum PlanNodeType {
return InsertTabletNode.deserializeFromWAL(stream);
case 14:
return InsertRowNode.deserializeFromWAL(stream);
+ case 44:
+ return DeleteDataNode.deserializeFromWAL(stream);
default:
throw new IllegalArgumentException("Invalid node type: " + nodeType);
}
@@ -173,6 +175,8 @@ public enum PlanNodeType {
return InsertTabletNode.deserializeFromWAL(buffer);
case 14:
return InsertRowNode.deserializeFromWAL(buffer);
+ case 44:
+ return DeleteDataNode.deserializeFromWAL(buffer);
default:
throw new IllegalArgumentException("Invalid node type: " + nodeType);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java
index 13e6294df3..022cd4f10b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.mpp.plan.planner.plan.node.write;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.metadata.path.PathDeserializeUtil;
@@ -31,8 +32,12 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.db.wal.buffer.IWALByteBufferView;
+import org.apache.iotdb.db.wal.buffer.WALEntryValue;
+import org.apache.iotdb.db.wal.utils.WALWriteUtils;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -44,13 +49,22 @@ import java.util.Map;
import java.util.stream.Collectors;
import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD;
+import static org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode.NO_CONSENSUS_INDEX;
-public class DeleteDataNode extends WritePlanNode {
+public class DeleteDataNode extends WritePlanNode implements WALEntryValue {
+ /** byte: type, integer: pathList.size(), long: deleteStartTime, deleteEndTime, searchIndex */
+ private static final int FIXED_SERIALIZED_SIZE = Short.BYTES + Integer.BYTES + Long.BYTES * 3;
private final List<PartialPath> pathList;
private final long deleteStartTime;
private final long deleteEndTime;
+ /**
+ * this index is used by wal search, its order should be protected by the upper layer, and the
+ * value should start from 1
+ */
+ protected long searchIndex = NO_CONSENSUS_INDEX;
+
private TRegionReplicaSet regionReplicaSet;
public DeleteDataNode(
@@ -86,6 +100,15 @@ public class DeleteDataNode extends WritePlanNode {
return deleteEndTime;
}
+ public long getSearchIndex() {
+ return searchIndex;
+ }
+
+ /** Search index should start from 1 */
+ public void setSearchIndex(long searchIndex) {
+ this.searchIndex = searchIndex;
+ }
+
@Override
public List<PlanNode> getChildren() {
return new ArrayList<>();
@@ -109,6 +132,67 @@ public class DeleteDataNode extends WritePlanNode {
return null;
}
+ @Override
+ public int serializedSize() {
+ int size = FIXED_SERIALIZED_SIZE;
+ for (PartialPath path : pathList) {
+ size += ReadWriteIOUtils.sizeToWrite(path.getFullPath());
+ }
+ return size;
+ }
+
+ @Override
+ public void serializeToWAL(IWALByteBufferView buffer) {
+ buffer.putShort(PlanNodeType.DELETE_DATA.getNodeType());
+ buffer.putLong(searchIndex);
+ buffer.putInt(pathList.size());
+ for (PartialPath path : pathList) {
+ WALWriteUtils.write(path.getFullPath(), buffer);
+ }
+ buffer.putLong(deleteStartTime);
+ buffer.putLong(deleteEndTime);
+ }
+
+ public static DeleteDataNode deserializeFromWAL(DataInputStream stream) throws IOException {
+ long searchIndex = stream.readLong();
+ int size = stream.readInt();
+ List<PartialPath> pathList = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ try {
+ pathList.add(new PartialPath(ReadWriteIOUtils.readString(stream)));
+ } catch (IllegalPathException e) {
+ throw new IllegalArgumentException("Cannot deserialize InsertRowNode", e);
+ }
+ }
+ long deleteStartTime = stream.readLong();
+ long deleteEndTime = stream.readLong();
+
+ DeleteDataNode deleteDataNode =
+ new DeleteDataNode(new PlanNodeId(""), pathList, deleteStartTime, deleteEndTime);
+ deleteDataNode.setSearchIndex(searchIndex);
+ return deleteDataNode;
+ }
+
+ public static DeleteDataNode deserializeFromWAL(ByteBuffer buffer) {
+ long searchIndex = buffer.getLong();
+ int size = buffer.getInt();
+ List<PartialPath> pathList = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ try {
+ pathList.add(new PartialPath(ReadWriteIOUtils.readString(buffer)));
+ } catch (IllegalPathException e) {
+ throw new IllegalArgumentException("Cannot deserialize InsertRowNode", e);
+ }
+ }
+ long deleteStartTime = buffer.getLong();
+ long deleteEndTime = buffer.getLong();
+
+ DeleteDataNode deleteDataNode =
+ new DeleteDataNode(new PlanNodeId(""), pathList, deleteStartTime, deleteEndTime);
+ deleteDataNode.setSearchIndex(searchIndex);
+ return deleteDataNode;
+ }
+
@Override
protected void serializeAttributes(ByteBuffer byteBuffer) {
PlanNodeType.DELETE_DATA.serialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java
index f5709b338d..7e3b811148 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.utils.MmapUtil;
import org.apache.iotdb.db.wal.exception.WALNodeClosedException;
@@ -234,12 +235,14 @@ public class WALBuffer extends AbstractWALBuffer {
}
// update search index
long searchIndex = DEFAULT_SEARCH_INDEX;
- if (walEntry.getType() == WALEntryType.INSERT_TABLET_NODE
- || walEntry.getType() == WALEntryType.INSERT_ROW_NODE) {
- InsertNode insertNode = (InsertNode) walEntry.getValue();
- if (insertNode.getSearchIndex() != InsertNode.NO_CONSENSUS_INDEX) {
- searchIndex = insertNode.getSearchIndex();
- currentSearchIndex = insertNode.getSearchIndex();
+ if (walEntry.getType().needSearch()) {
+ if (walEntry.getType() == WALEntryType.DELETE_DATA_NODE) {
+ searchIndex = ((DeleteDataNode) walEntry.getValue()).getSearchIndex();
+ } else {
+ searchIndex = ((InsertNode) walEntry.getValue()).getSearchIndex();
+ }
+ if (searchIndex != DEFAULT_SEARCH_INDEX) {
+ currentSearchIndex = searchIndex;
currentFileStatus = WALFileStatus.CONTAINS_SEARCH_INDEX;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALEntry.java b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALEntry.java
index 27ae7f867d..4286c9f386 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALEntry.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALEntry.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.engine.memtable.AbstractMemTable;
import org.apache.iotdb.db.engine.memtable.IMemTable;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
@@ -74,6 +75,8 @@ public abstract class WALEntry implements SerializedSize {
this.type = WALEntryType.INSERT_ROW_NODE;
} else if (value instanceof InsertTabletNode) {
this.type = WALEntryType.INSERT_TABLET_NODE;
+ } else if (value instanceof DeleteDataNode) {
+ this.type = WALEntryType.DELETE_DATA_NODE;
} else {
throw new RuntimeException("Unknown WALEntry type");
}
@@ -93,9 +96,6 @@ public abstract class WALEntry implements SerializedSize {
throws IllegalPathException, IOException {
byte typeNum = stream.readByte();
WALEntryType type = WALEntryType.valueOf(typeNum);
- if (type == null) {
- throw new IOException("unrecognized wal entry type " + typeNum);
- }
// handle signal
switch (type) {
@@ -127,6 +127,9 @@ public abstract class WALEntry implements SerializedSize {
case INSERT_TABLET_NODE:
value = (InsertTabletNode) PlanNodeType.deserializeFromWAL(stream);
break;
+ case DELETE_DATA_NODE:
+ value = (DeleteDataNode) PlanNodeType.deserializeFromWAL(stream);
+ break;
}
return new WALInfoEntry(type, memTableId, value);
}
@@ -135,7 +138,7 @@ public abstract class WALEntry implements SerializedSize {
* This deserialization method is only for multi-leader consensus and just deserializes
* InsertRowNode and InsertTabletNode
*/
- public static PlanNode deserializeInsertNode(ByteBuffer buffer) {
+ public static PlanNode deserializeForConsensus(ByteBuffer buffer) {
logger.debug(
"buffer capacity is: {}, limit is: {}, position is: {}",
buffer.capacity(),
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALEntryType.java b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALEntryType.java
index d6483db413..be32b339d7 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALEntryType.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALEntryType.java
@@ -33,6 +33,8 @@ public enum WALEntryType {
INSERT_ROW_NODE((byte) 4),
/** {@link org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode} */
INSERT_TABLET_NODE((byte) 5),
+ /** {@link org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode} */
+ DELETE_DATA_NODE((byte) 6),
// endregion
// region signal entry type
/** signal wal buffer has been closed */
@@ -54,12 +56,17 @@ public enum WALEntryType {
return code;
}
+ /** Returns true when this type should be searched */
+ public boolean needSearch() {
+ return this == INSERT_TABLET_NODE || this == INSERT_ROW_NODE || this == DELETE_DATA_NODE;
+ }
+
public static WALEntryType valueOf(byte code) {
for (WALEntryType type : WALEntryType.values()) {
if (type.code == code) {
return type;
}
}
- return null;
+ throw new IllegalArgumentException("Invalid WALEntryType code: " + code);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALInfoEntry.java b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALInfoEntry.java
index 58635400b7..a47eff9a54 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALInfoEntry.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALInfoEntry.java
@@ -82,6 +82,7 @@ public class WALInfoEntry extends WALEntry {
case INSERT_ROW_PLAN:
case INSERT_ROW_NODE:
case DELETE_PLAN:
+ case DELETE_DATA_NODE:
case MEMORY_TABLE_SNAPSHOT:
value.serializeToWAL(buffer);
break;
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/node/IWALNode.java b/server/src/main/java/org/apache/iotdb/db/wal/node/IWALNode.java
index a9958fffbf..38b12a8a85 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/node/IWALNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/node/IWALNode.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.consensus.multileader.wal.ConsensusReqReader;
import org.apache.iotdb.db.engine.flush.FlushListener;
import org.apache.iotdb.db.engine.memtable.IMemTable;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
@@ -46,6 +47,9 @@ public interface IWALNode extends FlushListener, AutoCloseable, ConsensusReqRead
/** Log DeletePlan */
WALFlushListener log(long memTableId, DeletePlan deletePlan);
+ /** Log DeleteDataNode */
+ WALFlushListener log(long memTableId, DeleteDataNode deleteDataNode);
+
/** Callback when memTable created */
void onMemTableCreated(IMemTable memTable, String targetTsFile);
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/node/WALFakeNode.java b/server/src/main/java/org/apache/iotdb/db/wal/node/WALFakeNode.java
index 40957b065c..e9cfb6d435 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/node/WALFakeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/node/WALFakeNode.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.wal.node;
import org.apache.iotdb.db.engine.memtable.IMemTable;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
@@ -68,6 +69,11 @@ public class WALFakeNode implements IWALNode {
return getResult();
}
+ @Override
+ public WALFlushListener log(long memTableId, DeleteDataNode deleteDataNode) {
+ return getResult();
+ }
+
private WALFlushListener getResult() {
WALFlushListener walFlushListener = new WALFlushListener(false);
switch (status) {
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/node/WALNode.java b/server/src/main/java/org/apache/iotdb/db/wal/node/WALNode.java
index ff42c5174c..174ede96f2 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/node/WALNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/node/WALNode.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.db.engine.memtable.IMemTable;
import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
@@ -162,6 +163,12 @@ public class WALNode implements IWALNode {
return log(walEntry);
}
+ @Override
+ public WALFlushListener log(long memTableId, DeleteDataNode deleteDataNode) {
+ WALEntry walEntry = new WALInfoEntry(memTableId, deleteDataNode);
+ return log(walEntry);
+ }
+
private WALFlushListener log(WALEntry walEntry) {
buffer.write(walEntry);
return walEntry.getWalFlushListener();
@@ -540,7 +547,7 @@ public class WALNode implements IWALNode {
while (walByteBufReader.hasNext()) {
ByteBuffer buffer = walByteBufReader.next();
WALEntryType type = WALEntryType.valueOf(buffer.get());
- if (type == WALEntryType.INSERT_TABLET_NODE || type == WALEntryType.INSERT_ROW_NODE) {
+ if (type.needSearch()) {
// see WALInfoEntry#serialize, entry type + memtable id + insert node type
buffer.position(WALInfoEntry.FIXED_SERIALIZED_SIZE + PlanNodeType.BYTES);
long searchIndex = buffer.getLong();
@@ -596,8 +603,7 @@ public class WALNode implements IWALNode {
while (walByteBufReader.hasNext()) {
ByteBuffer buffer = walByteBufReader.next();
WALEntryType type = WALEntryType.valueOf(buffer.get());
- if (type == WALEntryType.INSERT_TABLET_NODE
- || type == WALEntryType.INSERT_ROW_NODE) {
+ if (type.needSearch()) {
// see WALInfoEntry#serialize, entry type + memtable id + insert node type
buffer.position(WALInfoEntry.FIXED_SERIALIZED_SIZE + PlanNodeType.BYTES);
long searchIndex = buffer.getLong();
@@ -686,7 +692,10 @@ public class WALNode implements IWALNode {
boolean timeout =
!buffer.waitForFlush(WAIT_FOR_NEXT_WAL_ENTRY_TIMEOUT_IN_SEC, TimeUnit.SECONDS);
if (timeout) {
- logger.info("timeout when waiting for next WAL entry ready, execute rollWALFile");
+ logger.info(
+ "timeout when waiting for next WAL entry ready, execute rollWALFile. Current search index in wal buffer is {}, and next target index is {}",
+ buffer.getCurrentSearchIndex(),
+ nextSearchIndex);
rollWALFile();
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/recover/file/TsFilePlanRedoer.java b/server/src/main/java/org/apache/iotdb/db/wal/recover/file/TsFilePlanRedoer.java
index d69798fdca..eef04fe04e 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/recover/file/TsFilePlanRedoer.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/recover/file/TsFilePlanRedoer.java
@@ -34,6 +34,7 @@ import org.apache.iotdb.db.metadata.idtable.IDTable;
import org.apache.iotdb.db.metadata.idtable.entry.DeviceIDFactory;
import org.apache.iotdb.db.metadata.mnode.IMeasurementMNode;
import org.apache.iotdb.db.mpp.plan.analyze.SchemaValidator;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
@@ -89,6 +90,26 @@ public class TsFilePlanRedoer {
}
}
+ void redoDelete(DeleteDataNode deleteDataNode) throws IOException {
+ List<PartialPath> paths = deleteDataNode.getPathList();
+ for (PartialPath path : paths) {
+ // path here is device path pattern
+ recoveryMemTable.delete(
+ path,
+ path.getDevicePath(),
+ deleteDataNode.getDeleteStartTime(),
+ deleteDataNode.getDeleteEndTime());
+ tsFileResource
+ .getModFile()
+ .write(
+ new Deletion(
+ path,
+ tsFileResource.getTsFileSize(),
+ deleteDataNode.getDeleteStartTime(),
+ deleteDataNode.getDeleteEndTime()));
+ }
+ }
+
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
void redoInsert(InsertPlan plan) throws WriteProcessException, QueryProcessException {
if (!plan.hasValidMeasurements()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/recover/file/UnsealedTsFileRecoverPerformer.java b/server/src/main/java/org/apache/iotdb/db/wal/recover/file/UnsealedTsFileRecoverPerformer.java
index 926e54a75b..ab2e96ccd9 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/recover/file/UnsealedTsFileRecoverPerformer.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/recover/file/UnsealedTsFileRecoverPerformer.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.DataRegionException;
import org.apache.iotdb.db.metadata.idtable.IDTable;
import org.apache.iotdb.db.metadata.idtable.entry.IDeviceID;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
@@ -203,6 +204,9 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform
case INSERT_TABLET_NODE:
walRedoer.redoInsert((InsertNode) walEntry.getValue());
break;
+ case DELETE_DATA_NODE:
+ walRedoer.redoDelete((DeleteDataNode) walEntry.getValue());
+ break;
}
} catch (Exception e) {
logger.warn("meet error when redo wal of {}", tsFileResource.getTsFile(), e);
diff --git a/server/src/test/java/org/apache/iotdb/db/wal/io/WALFileTest.java b/server/src/test/java/org/apache/iotdb/db/wal/io/WALFileTest.java
index e6a8ecef46..7d02b1b90f 100644
--- a/server/src/test/java/org/apache/iotdb/db/wal/io/WALFileTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/wal/io/WALFileTest.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.constant.TestConstant;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
@@ -47,6 +48,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import static org.junit.Assert.assertEquals;
@@ -78,6 +80,7 @@ public class WALFileTest {
List<WALEntry> expectedWALEntries = new ArrayList<>();
expectedWALEntries.add(new WALInfoEntry(fakeMemTableId, getInsertRowNode(devicePath)));
expectedWALEntries.add(new WALInfoEntry(fakeMemTableId, getInsertTabletNode(devicePath)));
+ expectedWALEntries.add(new WALInfoEntry(fakeMemTableId, getDeleteDataNode(devicePath)));
expectedWALEntries.add(new WALInfoEntry(fakeMemTableId, getInsertRowPlan(devicePath)));
expectedWALEntries.add(new WALInfoEntry(fakeMemTableId, getInsertTabletPlan(devicePath)));
expectedWALEntries.add(new WALInfoEntry(fakeMemTableId, getDeletePlan(devicePath)));
@@ -323,4 +326,15 @@ public class WALFileTest {
public static DeletePlan getDeletePlan(String devicePath) throws IllegalPathException {
return new DeletePlan(Long.MIN_VALUE, Long.MAX_VALUE, new PartialPath(devicePath));
}
+
+ public static DeleteDataNode getDeleteDataNode(String devicePath) throws IllegalPathException {
+ DeleteDataNode deleteDataNode =
+ new DeleteDataNode(
+ new PlanNodeId(""),
+ Collections.singletonList(new PartialPath(devicePath)),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ deleteDataNode.setSearchIndex(100L);
+ return deleteDataNode;
+ }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/wal/node/ConsensusReqReaderTest.java b/server/src/test/java/org/apache/iotdb/db/wal/node/ConsensusReqReaderTest.java
index 1be5e36df1..9c59f58a26 100644
--- a/server/src/test/java/org/apache/iotdb/db/wal/node/ConsensusReqReaderTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/wal/node/ConsensusReqReaderTest.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.constant.TestConstant;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.utils.EnvironmentUtils;
@@ -43,6 +44,7 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -73,16 +75,9 @@ public class ConsensusReqReaderTest {
}
/**
- * Generate wal files as below: <br>
- * _0-0-1.wal: 1,-1 <br>
- * _1-1-1.wal: 2,2,2 <br>
- * _2-2-1.wal: 3,3 <br>
- * _3-3-1.wal: 3,4 <br>
- * _4-4-1.wal: 4 <br>
- * _5-4-1.wal: 4,4,5 <br>
- * _6-5-1.wal: 6 <br>
- * 1 - InsertRowNode, 2 - InsertRowsOfOneDeviceNode, 3 - InsertRowsNode, 4 -
- * InsertMultiTabletsNode, 5 - InsertTabletNode, 6 - InsertRowNode
+ * Generate wal files as below: _0-0-1.wal: 1,-1 _1-1-1.wal: 2,2,2 _2-2-1.wal: 3,3 _3-3-1.wal: 3,4
+ * _4-4-1.wal: 4 _5-4-1.wal: 4,4,5 _6-5-1.wal: 6 1 - InsertRowNode, 2 - InsertRowsOfOneDeviceNode,
+ * 3 - InsertRowsNode, 4 - InsertMultiTabletsNode, 5 - InsertTabletNode, 6 - InsertRowNode
*/
private void simulateFileScenario01() throws IllegalPathException {
InsertTabletNode insertTabletNode;
@@ -143,7 +138,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(1, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertRowNode);
Assert.assertEquals(1, ((InsertRowNode) planNode).getSearchIndex());
}
@@ -151,7 +146,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(3, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertRowNode);
Assert.assertEquals(2, ((InsertRowNode) planNode).getSearchIndex());
}
@@ -159,7 +154,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(3, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertRowNode);
Assert.assertEquals(3, ((InsertRowNode) planNode).getSearchIndex());
}
@@ -167,7 +162,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(4, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertTabletNode);
Assert.assertEquals(4, ((InsertTabletNode) planNode).getSearchIndex());
}
@@ -175,7 +170,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(1, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertTabletNode);
Assert.assertEquals(5, ((InsertTabletNode) planNode).getSearchIndex());
}
@@ -194,7 +189,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(4, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertTabletNode);
Assert.assertEquals(4, ((InsertTabletNode) planNode).getSearchIndex());
}
@@ -211,7 +206,7 @@ public class ConsensusReqReaderTest {
Assert.assertEquals(1, req.getRequests().size());
for (IConsensusRequest innerRequest : req.getRequests()) {
PlanNode node =
- WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(node instanceof InsertTabletNode);
Assert.assertEquals(5, ((InsertTabletNode) node).getSearchIndex());
}
@@ -244,7 +239,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(1, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertTabletNode);
Assert.assertEquals(5, ((InsertTabletNode) planNode).getSearchIndex());
}
@@ -253,7 +248,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(1, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertRowNode);
Assert.assertEquals(6, ((InsertRowNode) planNode).getSearchIndex());
}
@@ -283,7 +278,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(1, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertRowNode);
Assert.assertEquals(1, ((InsertRowNode) planNode).getSearchIndex());
}
@@ -291,7 +286,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(3, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertRowNode);
Assert.assertEquals(2, ((InsertRowNode) planNode).getSearchIndex());
}
@@ -302,7 +297,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(4, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertTabletNode);
Assert.assertEquals(4, ((InsertTabletNode) planNode).getSearchIndex());
}
@@ -310,7 +305,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(1, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertTabletNode);
Assert.assertEquals(5, ((InsertTabletNode) planNode).getSearchIndex());
}
@@ -330,7 +325,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(1, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertTabletNode);
Assert.assertEquals(5, ((InsertTabletNode) planNode).getSearchIndex());
}
@@ -341,7 +336,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(3, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertRowNode);
Assert.assertEquals(2, ((InsertRowNode) planNode).getSearchIndex());
}
@@ -349,7 +344,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(3, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertRowNode);
Assert.assertEquals(3, ((InsertRowNode) planNode).getSearchIndex());
}
@@ -357,7 +352,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(4, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertTabletNode);
Assert.assertEquals(4, ((InsertTabletNode) planNode).getSearchIndex());
}
@@ -365,7 +360,7 @@ public class ConsensusReqReaderTest {
request = iterator.next();
Assert.assertEquals(1, request.getRequests().size());
for (IConsensusRequest innerRequest : request.getRequests()) {
- planNode = WALEntry.deserializeInsertNode(innerRequest.serializeToByteBuffer());
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
Assert.assertTrue(planNode instanceof InsertTabletNode);
Assert.assertEquals(5, ((InsertTabletNode) planNode).getSearchIndex());
}
@@ -386,6 +381,46 @@ public class ConsensusReqReaderTest {
Assert.assertFalse(iterator.hasNext());
}
+ /**
+ * Generate wal files as below: _0-0-0.wal: -1,-1 _1-0-0.wal: -1 _2-0-1.wal: -1,1 _3-1-0.wal: -1 1
+ * - DeleteDataNode
+ */
+ private void simulateFileScenario02() throws IllegalPathException {
+ InsertRowNode insertRowNode = getInsertRowNode(devicePath);
+ // _0-0-0.wal
+ walNode.log(0, insertRowNode); // -1
+ walNode.log(0, insertRowNode); // -1
+ walNode.rollWALFile();
+ // _1-0-0.wal
+ walNode.log(0, insertRowNode); // -1
+ walNode.rollWALFile();
+ // _2-0-1.wal
+ walNode.log(0, insertRowNode); // -1
+ DeleteDataNode deleteDataNode = getDeleteDataNode(devicePath);
+ deleteDataNode.setSearchIndex(1);
+ walNode.log(0, deleteDataNode); // 1
+ walNode.rollWALFile();
+ // _3-1-0.wal
+ walNode.log(0, insertRowNode); // -1
+ walNode.rollWALFile();
+ }
+
+ @Test
+ public void scenario02TestGetReqIterator01() throws Exception {
+ simulateFileScenario02();
+ ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1);
+ IndexedConsensusRequest request;
+ PlanNode planNode;
+ Assert.assertTrue(iterator.hasNext());
+ request = iterator.next();
+ Assert.assertEquals(1, request.getRequests().size());
+ for (IConsensusRequest innerRequest : request.getRequests()) {
+ planNode = WALEntry.deserializeForConsensus(innerRequest.serializeToByteBuffer());
+ Assert.assertTrue(planNode instanceof DeleteDataNode);
+ Assert.assertEquals(1, ((DeleteDataNode) planNode).getSearchIndex());
+ }
+ }
+
public static InsertRowNode getInsertRowNode(String devicePath) throws IllegalPathException {
long time = 110L;
TSDataType[] dataTypes =
@@ -490,4 +525,12 @@ public class ConsensusReqReaderTest {
return insertTabletNode;
}
+
+ private DeleteDataNode getDeleteDataNode(String devicePath) throws IllegalPathException {
+ return new DeleteDataNode(
+ new PlanNodeId(""),
+ Collections.singletonList(new PartialPath(devicePath)),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/wal/recover/file/TsFilePlanRedoerTest.java b/server/src/test/java/org/apache/iotdb/db/wal/recover/file/TsFilePlanRedoerTest.java
index feab7d9d82..cbcb8c6b61 100644
--- a/server/src/test/java/org/apache/iotdb/db/wal/recover/file/TsFilePlanRedoerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/wal/recover/file/TsFilePlanRedoerTest.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.path.AlignedPath;
import org.apache.iotdb.db.metadata.path.MeasurementPath;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
@@ -135,6 +137,10 @@ public class TsFilePlanRedoerTest {
if (tsFileResource != null) {
tsFileResource.close();
}
+ File modsFile = new File(FILE_NAME.concat(ModificationFile.FILE_SUFFIX));
+ if (modsFile.exists()) {
+ modsFile.delete();
+ }
EnvironmentUtils.cleanEnv();
// reset config
IoTDBDescriptor.getInstance()
@@ -554,6 +560,33 @@ public class TsFilePlanRedoerTest {
assertTrue(modsFile.exists());
}
+ @Test
+ public void testRedoDeleteDataNode() throws Exception {
+ // generate .tsfile and update resource in memory
+ File file = new File(FILE_NAME);
+ generateCompleteFile(file);
+ tsFileResource = new TsFileResource(file);
+ tsFileResource.updateStartTime(DEVICE1_NAME, 1);
+ tsFileResource.updateEndTime(DEVICE1_NAME, 2);
+ tsFileResource.updateStartTime(DEVICE2_NAME, 3);
+ tsFileResource.updateEndTime(DEVICE2_NAME, 4);
+
+ // generate DeletePlan
+ DeleteDataNode deleteDataNode =
+ new DeleteDataNode(
+ new PlanNodeId(""),
+ Collections.singletonList(new PartialPath(DEVICE1_NAME)),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+
+ // redo DeletePlan, vsg processor is used to test IdTable, don't test IdTable here
+ File modsFile = new File(FILE_NAME.concat(ModificationFile.FILE_SUFFIX));
+ assertFalse(modsFile.exists());
+ TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource, false, null);
+ planRedoer.redoDelete(deleteDataNode);
+ assertTrue(modsFile.exists());
+ }
+
@Test
public void testRedoAlignedInsertAfterDeleteTimeseries() throws Exception {
// some timeseries have been deleted