You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/05/14 12:05:29 UTC
(iotdb) 01/01: Opti
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch wal-fix
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 9d210d472923cda6bfc7b22228848bd048af018d
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Thu May 14 20:04:53 2026 +0800
Opti
---
.../dataregion/DataExecutionVisitor.java | 5 -
.../planner/plan/node/write/InsertRowNode.java | 6 +-
.../planner/plan/node/write/InsertRowsNode.java | 6 +-
.../planner/plan/node/write/InsertTabletNode.java | 6 +-
.../plan/node/write/RelationalInsertRowNode.java | 4 +-
.../plan/node/write/RelationalInsertRowsNode.java | 4 +-
.../plan/planner/plan/node/write/SearchNode.java | 37 ++++++
.../db/storageengine/dataregion/DataRegion.java | 98 ++++++++++++--
.../storageengine/dataregion/wal/node/WALNode.java | 11 +-
.../planner/node/write/InsertRowNodeSerdeTest.java | 90 +++++++++++++
.../node/write/InsertRowsNodeSerdeTest.java | 141 +++++++++++++++++++++
.../node/write/InsertTabletNodeSerdeTest.java | 67 ++++++++++
.../wal/node/WALNodeWaitForRollFileTest.java | 30 ++++-
13 files changed, 470 insertions(+), 35 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java
index 5de040eeba2..2e9a2dd4871 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java
@@ -75,7 +75,6 @@ public class DataExecutionVisitor implements PlanVisitor<TSStatus, DataRegion> {
public TSStatus visitInsertRow(InsertRowNode node, DataRegion dataRegion) {
try {
dataRegion.insert(node);
- dataRegion.insertSeparatorToWAL();
return StatusUtils.OK;
} catch (OutOfTTLException e) {
LOGGER.warn("Error in executing plan node: {}, caused by {}", node, e.getMessage());
@@ -99,7 +98,6 @@ public class DataExecutionVisitor implements PlanVisitor<TSStatus, DataRegion> {
public TSStatus visitInsertTablet(final InsertTabletNode node, final DataRegion dataRegion) {
try {
dataRegion.insertTablet(node);
- dataRegion.insertSeparatorToWAL();
return StatusUtils.OK;
} catch (final OutOfTTLException e) {
LOGGER.debug("Error in executing plan node: {}, caused by {}", node, e.getMessage());
@@ -136,7 +134,6 @@ public class DataExecutionVisitor implements PlanVisitor<TSStatus, DataRegion> {
public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) {
try {
dataRegion.insert(node);
- dataRegion.insertSeparatorToWAL();
return StatusUtils.OK;
} catch (WriteProcessRejectException e) {
LOGGER.warn("Reject in executing plan node: {}, caused by {}", node, e.getMessage());
@@ -173,7 +170,6 @@ public class DataExecutionVisitor implements PlanVisitor<TSStatus, DataRegion> {
public TSStatus visitInsertMultiTablets(InsertMultiTabletsNode node, DataRegion dataRegion) {
try {
dataRegion.insertTablets(node);
- dataRegion.insertSeparatorToWAL();
return StatusUtils.OK;
} catch (WriteProcessRejectException e) {
LOGGER.warn("Reject in executing plan node: {}, caused by {}", node, e.getMessage());
@@ -208,7 +204,6 @@ public class DataExecutionVisitor implements PlanVisitor<TSStatus, DataRegion> {
InsertRowsOfOneDeviceNode node, DataRegion dataRegion) {
try {
dataRegion.insert(node);
- dataRegion.insertSeparatorToWAL();
return StatusUtils.OK;
} catch (WriteProcessRejectException e) {
LOGGER.warn("Reject in executing plan node: {}, caused by {}", node, e.getMessage());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java
index 290ec8a4a71..6bc91acd008 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java
@@ -619,7 +619,7 @@ public class InsertRowNode extends InsertNode implements WALEntryValue {
@Override
public void serializeToWAL(IWALByteBufferView buffer) {
buffer.putShort(getType().getNodeType());
- buffer.putLong(searchIndex);
+ buffer.putLong(getEncodedSearchIndex());
subSerialize(buffer);
}
@@ -700,7 +700,7 @@ public class InsertRowNode extends InsertNode implements WALEntryValue {
public static InsertRowNode deserializeFromWAL(DataInputStream stream) throws IOException {
long searchIndex = stream.readLong();
InsertRowNode insertNode = subDeserializeFromWAL(stream);
- insertNode.setSearchIndex(searchIndex);
+ insertNode.setSearchIndexFromWAL(searchIndex);
return insertNode;
}
@@ -791,7 +791,7 @@ public class InsertRowNode extends InsertNode implements WALEntryValue {
public static InsertRowNode deserializeFromWAL(ByteBuffer buffer) {
long searchIndex = buffer.getLong();
InsertRowNode insertNode = subDeserializeFromWAL(buffer);
- insertNode.setSearchIndex(searchIndex);
+ insertNode.setSearchIndexFromWAL(searchIndex);
return insertNode;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsNode.java
index 28f914e1606..2be39501781 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsNode.java
@@ -349,7 +349,7 @@ public class InsertRowsNode extends InsertNode implements WALEntryValue {
@Override
public void serializeToWAL(IWALByteBufferView buffer) {
buffer.putShort(getType().getNodeType());
- buffer.putLong(searchIndex);
+ buffer.putLong(getEncodedSearchIndex());
subSerialize(buffer);
}
@@ -377,7 +377,7 @@ public class InsertRowsNode extends InsertNode implements WALEntryValue {
InsertRowNode insertRowNode = InsertRowNode.subDeserializeFromWAL(stream);
insertRowsNode.addOneInsertRowNode(insertRowNode, i);
}
- insertRowsNode.setSearchIndex(searchIndex);
+ insertRowsNode.setSearchIndexFromWAL(searchIndex);
return insertRowsNode;
}
@@ -397,7 +397,7 @@ public class InsertRowsNode extends InsertNode implements WALEntryValue {
InsertRowNode insertRowNode = InsertRowNode.subDeserializeFromWAL(buffer);
insertRowsNode.addOneInsertRowNode(insertRowNode, i);
}
- insertRowsNode.setSearchIndex(searchIndex);
+ insertRowsNode.setSearchIndexFromWAL(searchIndex);
return insertRowsNode;
}
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 995e8a95e3f..25cf3c4fc7e 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
@@ -880,7 +880,7 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
}
void subSerialize(IWALByteBufferView buffer, List<int[]> rangeList) {
- buffer.putLong(searchIndex);
+ buffer.putLong(getEncodedSearchIndex());
WALWriteUtils.write(targetPath.getFullPath(), buffer);
// data types are serialized in measurement schemas
writeMeasurementSchemas(buffer);
@@ -1012,7 +1012,7 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
}
protected void subDeserializeFromWAL(DataInputStream stream) throws IOException {
- searchIndex = stream.readLong();
+ setSearchIndexFromWAL(stream.readLong());
try {
targetPath = readTargetPath(stream);
} catch (IllegalPathException e) {
@@ -1047,7 +1047,7 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
}
protected void subDeserializeFromWAL(ByteBuffer buffer) {
- searchIndex = buffer.getLong();
+ setSearchIndexFromWAL(buffer.getLong());
try {
targetPath = readTargetPath(buffer);
} catch (IllegalPathException e) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowNode.java
index e622dba30b9..8ef6802f047 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowNode.java
@@ -128,14 +128,14 @@ public class RelationalInsertRowNode extends InsertRowNode {
throws IOException {
long searchIndex = stream.readLong();
RelationalInsertRowNode insertNode = subDeserializeFromWAL(stream);
- insertNode.setSearchIndex(searchIndex);
+ insertNode.setSearchIndexFromWAL(searchIndex);
return insertNode;
}
public static RelationalInsertRowNode deserializeFromWAL(ByteBuffer buffer) {
long searchIndex = buffer.getLong();
RelationalInsertRowNode insertNode = subDeserializeFromWAL(buffer);
- insertNode.setSearchIndex(searchIndex);
+ insertNode.setSearchIndexFromWAL(searchIndex);
return insertNode;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowsNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowsNode.java
index 83498ceefc9..741c45f3256 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowsNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowsNode.java
@@ -124,7 +124,7 @@ public class RelationalInsertRowsNode extends InsertRowsNode {
RelationalInsertRowNode insertRowNode = RelationalInsertRowNode.subDeserializeFromWAL(stream);
insertRowsNode.addOneInsertRowNode(insertRowNode, i);
}
- insertRowsNode.setSearchIndex(searchIndex);
+ insertRowsNode.setSearchIndexFromWAL(searchIndex);
return insertRowsNode;
}
@@ -144,7 +144,7 @@ public class RelationalInsertRowsNode extends InsertRowsNode {
RelationalInsertRowNode insertRowNode = RelationalInsertRowNode.subDeserializeFromWAL(buffer);
insertRowsNode.addOneInsertRowNode(insertRowNode, i);
}
- insertRowsNode.setSearchIndex(searchIndex);
+ insertRowsNode.setSearchIndexFromWAL(searchIndex);
return insertRowsNode;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/SearchNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/SearchNode.java
index 5a7e5d9cf78..7f7e5bc9458 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/SearchNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/SearchNode.java
@@ -28,6 +28,8 @@ import java.util.List;
public abstract class SearchNode extends WritePlanNode implements ComparableConsensusRequest {
+ private static final long LAST_FRAGMENT_MASK = Long.MIN_VALUE;
+
/** this insert node doesn't need to participate in iot consensus */
public static final long NO_CONSENSUS_INDEX = ConsensusReqReader.DEFAULT_SEARCH_INDEX;
@@ -37,6 +39,8 @@ public abstract class SearchNode extends WritePlanNode implements ComparableCons
*/
protected long searchIndex = NO_CONSENSUS_INDEX;
+ protected boolean isLastFragment = false;
+
protected SearchNode(PlanNodeId id) {
super(id);
}
@@ -51,5 +55,38 @@ public abstract class SearchNode extends WritePlanNode implements ComparableCons
return this;
}
+ public boolean isLastFragment() {
+ return isLastFragment;
+ }
+
+ public SearchNode setLastFragment(boolean lastFragment) {
+ isLastFragment = lastFragment;
+ return this;
+ }
+
+ protected long getEncodedSearchIndex() {
+ if (searchIndex == NO_CONSENSUS_INDEX || !isLastFragment) {
+ return searchIndex;
+ }
+ return searchIndex | LAST_FRAGMENT_MASK;
+ }
+
+ public static long extractSearchIndex(long encodedSearchIndex) {
+ if (encodedSearchIndex == NO_CONSENSUS_INDEX) {
+ return encodedSearchIndex;
+ }
+ return encodedSearchIndex & ~LAST_FRAGMENT_MASK;
+ }
+
+ public static boolean isLastFragment(long encodedSearchIndex) {
+ return encodedSearchIndex != NO_CONSENSUS_INDEX
+ && (encodedSearchIndex & LAST_FRAGMENT_MASK) != 0;
+ }
+
+ protected void setSearchIndexFromWAL(long encodedSearchIndex) {
+ this.searchIndex = extractSearchIndex(encodedSearchIndex);
+ this.isLastFragment = isLastFragment(encodedSearchIndex);
+ }
+
public abstract SearchNode merge(List<SearchNode> searchNodes);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index a6808b8d972..a0e1732189e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -1194,6 +1194,7 @@ public class DataRegion implements IDataRegionForQuery {
// init map
long timePartitionId = TimePartitionUtils.getTimePartitionId(insertRowNode.getTime());
initFlushTimeMap(timePartitionId);
+ insertRowNode.setLastFragment(true);
boolean isSequence =
config.isEnableSeparateData()
@@ -1322,14 +1323,29 @@ public class DataRegion implements IDataRegionForQuery {
InsertTabletNode insertTabletNode,
Map<Long, List<int[]>[]> splitMap,
TSStatus[] results,
- long[] infoForMetrics)
+ long[] infoForMetrics,
+ boolean markLastFragmentOnFinalWrite)
throws DataTypeInconsistentException {
boolean noFailure = true;
+ int remainingFragmentCount = 0;
+ if (markLastFragmentOnFinalWrite) {
+ for (Entry<Long, List<int[]>[]> entry : splitMap.entrySet()) {
+ List<int[]>[] rangeLists = entry.getValue();
+ if (rangeLists[1] != null) {
+ remainingFragmentCount++;
+ }
+ if (rangeLists[0] != null) {
+ remainingFragmentCount++;
+ }
+ }
+ }
for (Entry<Long, List<int[]>[]> entry : splitMap.entrySet()) {
long timePartitionId = entry.getKey();
List<int[]>[] rangeLists = entry.getValue();
List<int[]> sequenceRangeList = rangeLists[1];
if (sequenceRangeList != null) {
+ insertTabletNode.setLastFragment(
+ markLastFragmentOnFinalWrite && remainingFragmentCount == 1);
noFailure =
insertTabletToTsFileProcessor(
insertTabletNode,
@@ -1340,9 +1356,12 @@ public class DataRegion implements IDataRegionForQuery {
noFailure,
infoForMetrics)
&& noFailure;
+ remainingFragmentCount--;
}
List<int[]> unSequenceRangeList = rangeLists[0];
if (unSequenceRangeList != null) {
+ insertTabletNode.setLastFragment(
+ markLastFragmentOnFinalWrite && remainingFragmentCount == 1);
noFailure =
insertTabletToTsFileProcessor(
insertTabletNode,
@@ -1353,6 +1372,7 @@ public class DataRegion implements IDataRegionForQuery {
noFailure,
infoForMetrics)
&& noFailure;
+ remainingFragmentCount--;
}
}
return noFailure;
@@ -1391,7 +1411,7 @@ public class DataRegion implements IDataRegionForQuery {
// infoForMetrics[2]: ScheduleWalTimeCost
// infoForMetrics[3]: ScheduleMemTableTimeCost
// infoForMetrics[4]: InsertedPointsNumber
- boolean noFailure = executeInsertTablet(insertTabletNode, results, infoForMetrics);
+ boolean noFailure = executeInsertTablet(insertTabletNode, results, infoForMetrics, true);
updateTsFileProcessorMetric(insertTabletNode, infoForMetrics);
if (!noFailure) {
@@ -1407,7 +1427,8 @@ public class DataRegion implements IDataRegionForQuery {
InsertTabletNode insertTabletNode,
TSStatus[] results,
long[] infoForMetrics,
- List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs) {
+ List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs,
+ boolean markLastFragmentOnFinalWrite) {
final int initialStart = start;
try {
Map<Long, List<int[]>[]> splitInfo = new HashMap<>();
@@ -1416,7 +1437,8 @@ public class DataRegion implements IDataRegionForQuery {
split(insertTabletNode, start, end, splitInfo);
start = end;
}
- return doInsert(insertTabletNode, splitInfo, results, infoForMetrics);
+ return doInsert(
+ insertTabletNode, splitInfo, results, infoForMetrics, markLastFragmentOnFinalWrite);
} catch (DataTypeInconsistentException e) {
// the exception will trigger a flush, which requires the flush time to be recalculated
start = initialStart;
@@ -1427,7 +1449,8 @@ public class DataRegion implements IDataRegionForQuery {
start = end;
}
try {
- return doInsert(insertTabletNode, splitInfo, results, infoForMetrics);
+ return doInsert(
+ insertTabletNode, splitInfo, results, infoForMetrics, markLastFragmentOnFinalWrite);
} catch (DataTypeInconsistentException ex) {
logger.error("Data inconsistent exception is not supposed to be triggered twice", ex);
return false;
@@ -1436,7 +1459,10 @@ public class DataRegion implements IDataRegionForQuery {
}
private boolean executeInsertTablet(
- InsertTabletNode insertTabletNode, TSStatus[] results, long[] infoForMetrics)
+ InsertTabletNode insertTabletNode,
+ TSStatus[] results,
+ long[] infoForMetrics,
+ boolean markLastFragmentOnFinalWrite)
throws OutOfTTLException {
boolean noFailure;
int loc =
@@ -1447,7 +1473,13 @@ public class DataRegion implements IDataRegionForQuery {
List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs =
insertTabletNode.splitByDevice(loc, insertTabletNode.getRowCount());
noFailure =
- splitAndInsert(loc, insertTabletNode, results, infoForMetrics, deviceEndOffsetPairs)
+ splitAndInsert(
+ loc,
+ insertTabletNode,
+ results,
+ infoForMetrics,
+ deviceEndOffsetPairs,
+ markLastFragmentOnFinalWrite)
&& noFailure;
if (CommonDescriptor.getInstance().getConfig().isLastCacheEnable()
@@ -1460,6 +1492,25 @@ public class DataRegion implements IDataRegionForQuery {
return noFailure;
}
+ private int findLastInsertTabletIndexToMark(final InsertMultiTabletsNode insertMultiTabletsNode) {
+ for (int i = insertMultiTabletsNode.getInsertTabletNodeList().size() - 1; i >= 0; i--) {
+ final InsertTabletNode insertTabletNode =
+ insertMultiTabletsNode.getInsertTabletNodeList().get(i);
+ if (insertTabletNode.getRowCount() <= 0 || insertTabletNode.allMeasurementFailed()) {
+ continue;
+ }
+ if (!insertTabletNode.shouldCheckTTL()) {
+ return i;
+ }
+ final long[] times = insertTabletNode.getTimes();
+ if (times.length > 0
+ && CommonUtils.isAlive(times[times.length - 1], getTTL(insertTabletNode))) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
private void initFlushTimeMap(long timePartitionId) {
if (config.isEnableSeparateData()
&& !lastFlushTimeMap.checkAndCreateFlushedTimePartition(timePartitionId, true)) {
@@ -1741,7 +1792,10 @@ public class DataRegion implements IDataRegionForQuery {
}
List<InsertRowNode> executedInsertRowNodeList = new ArrayList<>();
- for (Map.Entry<TsFileProcessor, InsertRowsNode> entry : tsFileProcessorMap.entrySet()) {
+ final List<Map.Entry<TsFileProcessor, InsertRowsNode>> groupedEntries =
+ new ArrayList<>(tsFileProcessorMap.entrySet());
+ markOnlyLastInsertFragment(groupedEntries, Map.Entry::getValue);
+ for (Map.Entry<TsFileProcessor, InsertRowsNode> entry : groupedEntries) {
InsertRowsNode subInsertRowsNode = entry.getValue();
try {
List<TsFileProcessor> insertedProcessors =
@@ -1760,6 +1814,13 @@ public class DataRegion implements IDataRegionForQuery {
return executedInsertRowNodeList;
}
+ private <T> void markOnlyLastInsertFragment(
+ final List<T> fragments, final java.util.function.Function<T, InsertNode> fragmentExtractor) {
+ for (int i = 0; i < fragments.size(); i++) {
+ fragmentExtractor.apply(fragments.get(i)).setLastFragment(i == fragments.size() - 1);
+ }
+ }
+
private List<TsFileProcessor> insertRowsWithTypeConsistencyCheck(
TsFileProcessor tsFileProcessor, InsertRowsNode subInsertRowsNode, long[] infoForMetrics)
throws WriteProcessException {
@@ -1853,8 +1914,17 @@ public class DataRegion implements IDataRegionForQuery {
});
}
+ final List<Entry<TsFileProcessor, InsertRowsNode>> retriedEntries =
+ new ArrayList<>(retriedProcessorMap.entrySet());
+ for (int i = 0; i < retriedEntries.size(); i++) {
+ retriedEntries
+ .get(i)
+ .getValue()
+ .setLastFragment(subInsertRowsNode.isLastFragment() && i == retriedEntries.size() - 1);
+ }
+
final List<TsFileProcessor> insertedProcessors = new ArrayList<>(retriedProcessorMap.size());
- for (Entry<TsFileProcessor, InsertRowsNode> retriedEntry : retriedProcessorMap.entrySet()) {
+ for (Entry<TsFileProcessor, InsertRowsNode> retriedEntry : retriedEntries) {
final TsFileProcessor retriedProcessor = retriedEntry.getKey();
registerToTsFile(retriedEntry.getValue(), retriedProcessor);
retriedProcessor.insertRows(retriedEntry.getValue(), infoForMetrics);
@@ -4587,7 +4657,10 @@ public class DataRegion implements IDataRegionForQuery {
// infoForMetrics[2]: ScheduleWalTimeCost
// infoForMetrics[3]: ScheduleMemTableTimeCost
// infoForMetrics[4]: InsertedPointsNumber
- for (Map.Entry<TsFileProcessor, InsertRowsNode> entry : tsFileProcessorMap.entrySet()) {
+ final List<Map.Entry<TsFileProcessor, InsertRowsNode>> groupedEntries =
+ new ArrayList<>(tsFileProcessorMap.entrySet());
+ markOnlyLastInsertFragment(groupedEntries, Map.Entry::getValue);
+ for (Map.Entry<TsFileProcessor, InsertRowsNode> entry : groupedEntries) {
InsertRowsNode subInsertRowsNode = entry.getValue();
try {
List<TsFileProcessor> insertedProcessors =
@@ -4728,6 +4801,7 @@ public class DataRegion implements IDataRegionForQuery {
// infoForMetrics[2]: ScheduleWalTimeCost
// infoForMetrics[3]: ScheduleMemTableTimeCost
// infoForMetrics[4]: InsertedPointsNumbe
+ final int lastTabletIndexToMark = findLastInsertTabletIndexToMark(insertMultiTabletsNode);
for (int i = 0; i < insertMultiTabletsNode.getInsertTabletNodeList().size(); i++) {
InsertTabletNode insertTabletNode = insertMultiTabletsNode.getInsertTabletNodeList().get(i);
long[] times = insertTabletNode.getTimes();
@@ -4741,7 +4815,9 @@ public class DataRegion implements IDataRegionForQuery {
Arrays.fill(results, RpcUtils.SUCCESS_STATUS);
boolean noFailure = false;
try {
- noFailure = executeInsertTablet(insertTabletNode, results, infoForMetrics);
+ noFailure =
+ executeInsertTablet(
+ insertTabletNode, results, infoForMetrics, i == lastTabletIndexToMark);
} catch (WriteProcessException e) {
insertMultiTabletsNode
.getResults()
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
index 98993c563f9..3a19909a6d6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
@@ -35,6 +35,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNo
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode;
import org.apache.iotdb.db.service.metrics.WritingMetrics;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
@@ -687,7 +688,9 @@ public class WALNode implements IWALNode {
if (type.needSearch()) {
// see WALInfoEntry#serialize, entry type + memtable id + plan node type
buffer.position(WALInfoEntry.FIXED_SERIALIZED_SIZE + PlanNodeType.BYTES);
- final long currentWalEntryIndex = buffer.getLong();
+ final long encodedSearchIndex = buffer.getLong();
+ final long currentWalEntryIndex = SearchNode.extractSearchIndex(encodedSearchIndex);
+ final boolean isLastFragment = SearchNode.isLastFragment(encodedSearchIndex);
buffer.clear();
if (currentWalEntryIndex == -1) {
// WAL entry of targetIndex has been fully collected, so put them into insertNodes
@@ -712,6 +715,9 @@ public class WALNode implements IWALNode {
tmpNodes.get().add(new IoTConsensusRequest(buffer));
memorySize += buffer.remaining();
}
+ if (isLastFragment) {
+ tryToCollectInsertNodeAndBumpIndex.run();
+ }
} else {
// currentWalEntryIndex > targetIndex
// WAL entry of targetIndex has been fully collected, put them into insertNodes
@@ -740,6 +746,9 @@ public class WALNode implements IWALNode {
tmpNodes.get().add(new IoTConsensusRequest(buffer));
memorySize += buffer.remaining();
}
+ if (isLastFragment) {
+ tryToCollectInsertNodeAndBumpIndex.run();
+ }
}
} else {
tryToCollectInsertNodeAndBumpIndex.run();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowNodeSerdeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowNodeSerdeTest.java
index dbe69da2462..21bc9e0d120 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowNodeSerdeTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowNodeSerdeTest.java
@@ -23,10 +23,13 @@ import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
+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.RelationalInsertRowNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.Assert;
import org.junit.Test;
@@ -35,6 +38,7 @@ import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
public class InsertRowNodeSerdeTest {
@@ -72,6 +76,7 @@ public class InsertRowNodeSerdeTest {
@Test
public void TestSerializeAndDeserializeForWAL() throws IllegalPathException, IOException {
InsertRowNode insertRowNode = getInsertRowNodeWithMeasurementSchemas();
+ insertRowNode.setLastFragment(true);
int serializedSize = insertRowNode.serializedSize();
@@ -96,6 +101,67 @@ public class InsertRowNodeSerdeTest {
new MeasurementSchema("s5", TSDataType.BOOLEAN)
});
Assert.assertEquals(insertRowNode, tmpNode);
+ Assert.assertTrue(tmpNode.isLastFragment());
+ }
+
+ @Test
+ public void testDeserializeLegacyWAL() throws IllegalPathException, IOException {
+ InsertRowNode insertRowNode = getInsertRowNodeWithMeasurementSchemas();
+ insertRowNode.setSearchIndex(123L);
+
+ byte[] bytes = new byte[insertRowNode.serializedSize()];
+ WALByteBufferForTest walBuffer = new WALByteBufferForTest(ByteBuffer.wrap(bytes));
+ insertRowNode.serializeToWAL(walBuffer);
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
+ Assert.assertEquals(PlanNodeType.INSERT_ROW.getNodeType(), byteBuffer.getShort());
+ Assert.assertEquals(123L, byteBuffer.getLong());
+
+ DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(bytes));
+ dataInputStream.readShort();
+
+ InsertRowNode tmpNode = InsertRowNode.deserializeFromWAL(dataInputStream);
+ tmpNode.setPlanNodeId(insertRowNode.getPlanNodeId());
+ tmpNode.setMeasurementSchemas(
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.DOUBLE),
+ new MeasurementSchema("s2", TSDataType.FLOAT),
+ new MeasurementSchema("s3", TSDataType.INT64),
+ new MeasurementSchema("s4", TSDataType.INT32),
+ new MeasurementSchema("s5", TSDataType.BOOLEAN)
+ });
+ Assert.assertEquals(insertRowNode, tmpNode);
+ Assert.assertEquals(123L, tmpNode.getSearchIndex());
+ Assert.assertFalse(tmpNode.isLastFragment());
+ }
+
+ @Test
+ public void testDeserializeLegacyWALRelational() throws IOException {
+ RelationalInsertRowNode insertRowNode = getRelationalInsertRowNodeWithMeasurementSchemas();
+ insertRowNode.setSearchIndex(123L);
+
+ byte[] bytes = new byte[insertRowNode.serializedSize()];
+ WALByteBufferForTest walBuffer = new WALByteBufferForTest(ByteBuffer.wrap(bytes));
+ insertRowNode.serializeToWAL(walBuffer);
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
+ Assert.assertEquals(PlanNodeType.RELATIONAL_INSERT_ROW.getNodeType(), byteBuffer.getShort());
+ Assert.assertEquals(123L, byteBuffer.getLong());
+
+ DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(bytes));
+ dataInputStream.readShort();
+
+ RelationalInsertRowNode tmpNode = RelationalInsertRowNode.deserializeFromWAL(dataInputStream);
+ tmpNode.setPlanNodeId(insertRowNode.getPlanNodeId());
+ tmpNode.setMeasurementSchemas(
+ new MeasurementSchema[] {
+ new MeasurementSchema("id", TSDataType.STRING),
+ new MeasurementSchema("attr", TSDataType.TEXT),
+ new MeasurementSchema("value", TSDataType.INT64)
+ });
+ Assert.assertEquals(insertRowNode, tmpNode);
+ Assert.assertEquals(123L, tmpNode.getSearchIndex());
+ Assert.assertFalse(tmpNode.isLastFragment());
}
private InsertRowNode getInsertRowNode() throws IllegalPathException {
@@ -199,4 +265,28 @@ public class InsertRowNodeSerdeTest {
insertRowNode.setNeedInferType(true);
return insertRowNode;
}
+
+ private RelationalInsertRowNode getRelationalInsertRowNodeWithMeasurementSchemas() {
+ return new RelationalInsertRowNode(
+ new PlanNodeId("plannode 3"),
+ new PartialPath("table1", false),
+ false,
+ new String[] {"id", "attr", "value"},
+ new TSDataType[] {TSDataType.STRING, TSDataType.TEXT, TSDataType.INT64},
+ new MeasurementSchema[] {
+ new MeasurementSchema("id", TSDataType.STRING),
+ new MeasurementSchema("attr", TSDataType.TEXT),
+ new MeasurementSchema("value", TSDataType.INT64)
+ },
+ 90L,
+ new Object[] {
+ new Binary("d1".getBytes(StandardCharsets.UTF_8)),
+ new Binary("v1".getBytes(StandardCharsets.UTF_8)),
+ 1L
+ },
+ false,
+ new TsTableColumnCategory[] {
+ TsTableColumnCategory.TAG, TsTableColumnCategory.ATTRIBUTE, TsTableColumnCategory.FIELD
+ });
+ }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowsNodeSerdeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowsNodeSerdeTest.java
index 08819830557..034d4cb57de 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowsNodeSerdeTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowsNodeSerdeTest.java
@@ -129,6 +129,7 @@ public class InsertRowsNodeSerdeTest {
new Object[] {2.0, false},
false),
1);
+ insertRowsNode.setLastFragment(true);
int serializedSize = insertRowsNode.serializedSize();
@@ -145,6 +146,70 @@ public class InsertRowsNodeSerdeTest {
InsertRowsNode tmpNode = InsertRowsNode.deserializeFromWAL(dataInputStream);
tmpNode.setPlanNodeId(insertRowsNode.getPlanNodeId());
Assert.assertEquals(insertRowsNode, tmpNode);
+ Assert.assertTrue(tmpNode.isLastFragment());
+ }
+
+ @Test
+ public void testDeserializeLegacyWAL() throws IllegalPathException, IOException {
+ InsertRowsNode insertRowsNode = new InsertRowsNode(new PlanNodeId("plan node 1"));
+ insertRowsNode.addOneInsertRowNode(
+ new InsertRowNode(
+ new PlanNodeId(""),
+ new PartialPath("root.sg.d1"),
+ false,
+ new String[] {"s1", "s2", "s3", "s4", "s5"},
+ new TSDataType[] {
+ TSDataType.DOUBLE,
+ TSDataType.FLOAT,
+ TSDataType.INT64,
+ TSDataType.TEXT,
+ TSDataType.STRING
+ },
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.DOUBLE),
+ new MeasurementSchema("s2", TSDataType.FLOAT),
+ new MeasurementSchema("s3", TSDataType.INT64),
+ new MeasurementSchema("s4", TSDataType.TEXT),
+ new MeasurementSchema("s5", TSDataType.STRING)
+ },
+ 1000L,
+ new Object[] {1.0, 2f, 300L, new Binary("444".getBytes(StandardCharsets.UTF_8)), null},
+ false),
+ 0);
+
+ insertRowsNode.addOneInsertRowNode(
+ new InsertRowNode(
+ new PlanNodeId(""),
+ new PartialPath("root.sg.d2"),
+ false,
+ new String[] {"s1", "s4"},
+ new TSDataType[] {TSDataType.DOUBLE, TSDataType.BOOLEAN},
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.DOUBLE),
+ new MeasurementSchema("s4", TSDataType.BOOLEAN),
+ },
+ 2000L,
+ new Object[] {2.0, false},
+ false),
+ 1);
+ insertRowsNode.setSearchIndex(123L);
+
+ byte[] bytes = new byte[insertRowsNode.serializedSize()];
+ WALByteBufferForTest walBuffer = new WALByteBufferForTest(ByteBuffer.wrap(bytes));
+ insertRowsNode.serializeToWAL(walBuffer);
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
+ Assert.assertEquals(PlanNodeType.INSERT_ROWS.getNodeType(), byteBuffer.getShort());
+ Assert.assertEquals(123L, byteBuffer.getLong());
+
+ DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(bytes));
+ dataInputStream.readShort();
+
+ InsertRowsNode tmpNode = InsertRowsNode.deserializeFromWAL(dataInputStream);
+ tmpNode.setPlanNodeId(insertRowsNode.getPlanNodeId());
+ Assert.assertEquals(insertRowsNode, tmpNode);
+ Assert.assertEquals(123L, tmpNode.getSearchIndex());
+ Assert.assertFalse(tmpNode.isLastFragment());
}
@Test
@@ -262,6 +327,7 @@ public class InsertRowsNodeSerdeTest {
TsTableColumnCategory.TAG, TsTableColumnCategory.ATTRIBUTE
}),
1);
+ insertRowsNode.setLastFragment(true);
int serializedSize = insertRowsNode.serializedSize();
@@ -280,6 +346,81 @@ public class InsertRowsNodeSerdeTest {
RelationalInsertRowsNode.deserializeFromWAL(dataInputStream);
tmpNode.setPlanNodeId(insertRowsNode.getPlanNodeId());
Assert.assertEquals(insertRowsNode, tmpNode);
+ Assert.assertTrue(tmpNode.isLastFragment());
}
}
+
+ @Test
+ public void testDeserializeLegacyWALRelational() throws IOException {
+ RelationalInsertRowsNode insertRowsNode =
+ new RelationalInsertRowsNode(new PlanNodeId("plan node 1"));
+ insertRowsNode.addOneInsertRowNode(
+ new RelationalInsertRowNode(
+ new PlanNodeId(""),
+ new PartialPath("table1", false),
+ false,
+ new String[] {"s1", "s2", "s3", "s4", "s5"},
+ new TSDataType[] {
+ TSDataType.DOUBLE,
+ TSDataType.FLOAT,
+ TSDataType.INT64,
+ TSDataType.TEXT,
+ TSDataType.STRING
+ },
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.DOUBLE),
+ new MeasurementSchema("s2", TSDataType.FLOAT),
+ new MeasurementSchema("s3", TSDataType.INT64),
+ new MeasurementSchema("s4", TSDataType.TEXT),
+ new MeasurementSchema("s5", TSDataType.STRING)
+ },
+ 1000L,
+ new Object[] {1.0, 2f, 300L, new Binary("444".getBytes(StandardCharsets.UTF_8)), null},
+ false,
+ new TsTableColumnCategory[] {
+ TsTableColumnCategory.TAG,
+ TsTableColumnCategory.ATTRIBUTE,
+ TsTableColumnCategory.FIELD,
+ TsTableColumnCategory.FIELD,
+ TsTableColumnCategory.FIELD
+ }),
+ 0);
+
+ insertRowsNode.addOneInsertRowNode(
+ new RelationalInsertRowNode(
+ new PlanNodeId(""),
+ new PartialPath("table1", false),
+ false,
+ new String[] {"s1", "s4"},
+ new TSDataType[] {TSDataType.DOUBLE, TSDataType.BOOLEAN},
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.DOUBLE),
+ new MeasurementSchema("s4", TSDataType.BOOLEAN),
+ },
+ 2000L,
+ new Object[] {2.0, false},
+ false,
+ new TsTableColumnCategory[] {
+ TsTableColumnCategory.TAG, TsTableColumnCategory.ATTRIBUTE
+ }),
+ 1);
+ insertRowsNode.setSearchIndex(123L);
+
+ byte[] bytes = new byte[insertRowsNode.serializedSize()];
+ WALByteBufferForTest walBuffer = new WALByteBufferForTest(ByteBuffer.wrap(bytes));
+ insertRowsNode.serializeToWAL(walBuffer);
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
+ Assert.assertEquals(PlanNodeType.RELATIONAL_INSERT_ROWS.getNodeType(), byteBuffer.getShort());
+ Assert.assertEquals(123L, byteBuffer.getLong());
+
+ DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(bytes));
+ dataInputStream.readShort();
+
+ RelationalInsertRowsNode tmpNode = RelationalInsertRowsNode.deserializeFromWAL(dataInputStream);
+ tmpNode.setPlanNodeId(insertRowsNode.getPlanNodeId());
+ Assert.assertEquals(insertRowsNode, tmpNode);
+ Assert.assertEquals(123L, tmpNode.getSearchIndex());
+ Assert.assertFalse(tmpNode.isLastFragment());
+ }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertTabletNodeSerdeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertTabletNodeSerdeTest.java
index 4d8f5f6e4a8..ddc35e1eda9 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertTabletNodeSerdeTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertTabletNodeSerdeTest.java
@@ -68,6 +68,7 @@ public class InsertTabletNodeSerdeTest {
@Test
public void testSerializeAndDeserializeForWAL() throws IllegalPathException, IOException {
InsertTabletNode insertTabletNode = getInsertTabletNodeWithSchema();
+ insertTabletNode.setLastFragment(true);
int serializedSize = insertTabletNode.serializedSize();
@@ -93,6 +94,38 @@ public class InsertTabletNodeSerdeTest {
new MeasurementSchema("s5", TSDataType.BOOLEAN)
});
Assert.assertEquals(insertTabletNode, tmpNode);
+ Assert.assertTrue(tmpNode.isLastFragment());
+ }
+
+ @Test
+ public void testDeserializeLegacyWAL() throws IllegalPathException, IOException {
+ InsertTabletNode insertTabletNode = getInsertTabletNodeWithSchema();
+ insertTabletNode.setSearchIndex(123L);
+
+ byte[] bytes = new byte[insertTabletNode.serializedSize()];
+ WALByteBufferForTest walBuffer = new WALByteBufferForTest(ByteBuffer.wrap(bytes));
+ insertTabletNode.serializeToWAL(walBuffer);
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
+ Assert.assertEquals(PlanNodeType.INSERT_TABLET.getNodeType(), byteBuffer.getShort());
+ Assert.assertEquals(123L, byteBuffer.getLong());
+
+ DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(bytes));
+ dataInputStream.readShort();
+
+ InsertTabletNode tmpNode = InsertTabletNode.deserializeFromWAL(dataInputStream);
+ tmpNode.setPlanNodeId(insertTabletNode.getPlanNodeId());
+ tmpNode.setMeasurementSchemas(
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.DOUBLE),
+ new MeasurementSchema("s2", TSDataType.FLOAT),
+ new MeasurementSchema("s3", TSDataType.INT64),
+ new MeasurementSchema("s4", TSDataType.INT32),
+ new MeasurementSchema("s5", TSDataType.BOOLEAN)
+ });
+ Assert.assertEquals(insertTabletNode, tmpNode);
+ Assert.assertEquals(123L, tmpNode.getSearchIndex());
+ Assert.assertFalse(tmpNode.isLastFragment());
}
@Test
@@ -126,6 +159,7 @@ public class InsertTabletNodeSerdeTest {
for (String tableName : new String[] {"table1", "ta`ble1", "root.table1"}) {
RelationalInsertTabletNode insertTabletNode =
getRelationalInsertTabletNodeWithSchema(tableName);
+ insertTabletNode.setLastFragment(true);
int serializedSize = insertTabletNode.serializedSize();
@@ -153,9 +187,42 @@ public class InsertTabletNodeSerdeTest {
new MeasurementSchema("s5", TSDataType.BOOLEAN)
});
Assert.assertEquals(insertTabletNode, tmpNode);
+ Assert.assertTrue(tmpNode.isLastFragment());
}
}
+ @Test
+ public void testDeserializeLegacyWALRelational() throws IOException {
+ RelationalInsertTabletNode insertTabletNode = getRelationalInsertTabletNodeWithSchema("table1");
+ insertTabletNode.setSearchIndex(123L);
+
+ byte[] bytes = new byte[insertTabletNode.serializedSize()];
+ WALByteBufferForTest walBuffer = new WALByteBufferForTest(ByteBuffer.wrap(bytes));
+ insertTabletNode.serializeToWAL(walBuffer);
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
+ Assert.assertEquals(PlanNodeType.RELATIONAL_INSERT_TABLET.getNodeType(), byteBuffer.getShort());
+ Assert.assertEquals(123L, byteBuffer.getLong());
+
+ DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(bytes));
+ dataInputStream.readShort();
+
+ RelationalInsertTabletNode tmpNode =
+ RelationalInsertTabletNode.deserializeFromWAL(dataInputStream);
+ tmpNode.setPlanNodeId(insertTabletNode.getPlanNodeId());
+ tmpNode.setMeasurementSchemas(
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.DOUBLE),
+ new MeasurementSchema("s2", TSDataType.FLOAT),
+ new MeasurementSchema("s3", TSDataType.INT64),
+ new MeasurementSchema("s4", TSDataType.INT32),
+ new MeasurementSchema("s5", TSDataType.BOOLEAN)
+ });
+ Assert.assertEquals(insertTabletNode, tmpNode);
+ Assert.assertEquals(123L, tmpNode.getSearchIndex());
+ Assert.assertFalse(tmpNode.isLastFragment());
+ }
+
@Test
public void testInitTabletValuesWithAllTypes()
throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java
index 2977841d5e6..c204a839b51 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java
@@ -20,7 +20,6 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.node;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
import org.apache.iotdb.db.conf.IoTDBConfig;
@@ -178,6 +177,29 @@ public class WALNodeWaitForRollFileTest {
assertNotNull(iterator.next());
}
+ @Test
+ public void testLegacySeparatorStillWorksAfterRollFile() throws Exception {
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
+ walNode.onMemTableCreated(memTable, logDirectory + File.separator + "test.tsfile");
+
+ InsertTabletNode insertTabletNode = getInsertTabletNode(devicePath, new long[] {1});
+ insertTabletNode.setSearchIndex(1);
+ walNode.log(
+ memTable.getMemTableId(),
+ insertTabletNode,
+ Collections.singletonList(new int[] {0, insertTabletNode.getRowCount()}));
+ walNode.log(memTable.getMemTableId(), new ContinuousSameSearchIndexSeparatorNode());
+
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> walNode.isAllWALEntriesConsumed());
+
+ walNode.rollWALFile();
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> walNode.isAllWALEntriesConsumed());
+
+ ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1);
+ assertTrue(iterator.hasNext());
+ assertNotNull(iterator.next());
+ }
+
/**
* Verifies that waitForNextReady wakes up when a WAL file roll is triggered concurrently. A
* background thread rolls the WAL file while the main thread waits on the iterator.
@@ -190,12 +212,11 @@ public class WALNodeWaitForRollFileTest {
// write data with search index
InsertTabletNode insertTabletNode = getInsertTabletNode(devicePath, new long[] {1});
insertTabletNode.setSearchIndex(1);
+ insertTabletNode.setLastFragment(true);
walNode.log(
memTable.getMemTableId(),
insertTabletNode,
Collections.singletonList(new int[] {0, insertTabletNode.getRowCount()}));
- walNode.log(
- memTable.getMemTableId(), new ContinuousSameSearchIndexSeparatorNode(new PlanNodeId("")));
Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> walNode.isAllWALEntriesConsumed());
@@ -318,12 +339,11 @@ public class WALNodeWaitForRollFileTest {
// write data with search index — stays in the current (active) WAL file
InsertTabletNode insertTabletNode = getInsertTabletNode(devicePath, new long[] {1});
insertTabletNode.setSearchIndex(1);
+ insertTabletNode.setLastFragment(true);
walNode.log(
memTable.getMemTableId(),
insertTabletNode,
Collections.singletonList(new int[] {0, insertTabletNode.getRowCount()}));
- walNode.log(
- memTable.getMemTableId(), new ContinuousSameSearchIndexSeparatorNode(new PlanNodeId("")));
Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> walNode.isAllWALEntriesConsumed());