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 2024/10/19 15:16:16 UTC
(iotdb) branch batch_wal_for_insertRelationalTablet created (now 5543d250797)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a change to branch batch_wal_for_insertRelationalTablet
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 5543d250797 Merge branch 'master' of github.com:apache/iotdb into batch_wal_for_insertRelationalTablet
This branch includes the following new commits:
new e8828b9dea3 try dev
new d1100baf800 dev
new d221dec6a87 fix compile
new 5543d250797 Merge branch 'master' of github.com:apache/iotdb into batch_wal_for_insertRelationalTablet
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 02/04: dev
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch batch_wal_for_insertRelationalTablet
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d1100baf8004d14d419a20f0bf3b45f860509e93
Author: HTHou <hh...@outlook.com>
AuthorDate: Sat Oct 19 17:41:11 2024 +0800
dev
---
.../apache/iotdb/db/storageengine/dataregion/DataRegion.java | 10 ----------
1 file changed, 10 deletions(-)
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 fba39e46471..25a8e570d64 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
@@ -1309,16 +1309,6 @@ public class DataRegion implements IDataRegionForQuery {
long timePartitionId,
boolean noFailure,
long[] costsForMetrics) {
- // return when start >= end or all measurement failed
- // if (start >= end || insertTabletNode.allMeasurementFailed()) {
- // if (logger.isDebugEnabled()) {
- // logger.debug(
- // "Won't insert tablet {}, because {}",
- // insertTabletNode.getSearchIndex(),
- // start >= end ? "start >= end" : "insertTabletNode allMeasurementFailed");
- // }
- // return true;
- // }
if (insertTabletNode.allMeasurementFailed()) {
if (logger.isDebugEnabled()) {
logger.debug(
(iotdb) 03/04: fix compile
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch batch_wal_for_insertRelationalTablet
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d221dec6a8721847f028715265168363c969d7e4
Author: HTHou <hh...@outlook.com>
AuthorDate: Sat Oct 19 23:15:32 2024 +0800
fix compile
---
.../dataregion/wal/buffer/WALInfoEntry.java | 2 +-
.../wal/node/ConsensusReqReaderTest.java | 31 +++++++++++++++++-----
.../dataregion/wal/node/WALFakeNodeTest.java | 13 +++++++--
.../dataregion/wal/node/WALNodeTest.java | 11 ++++++--
4 files changed, 46 insertions(+), 11 deletions(-)
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 08239c1f329..51d1a5e79b1 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
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.buffer;
-import java.util.ArrayList;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
@@ -27,6 +26,7 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode;
import org.apache.tsfile.utils.Pair;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/ConsensusReqReaderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/ConsensusReqReaderTest.java
index ca102ef8cf5..f7518cd109d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/ConsensusReqReaderTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/ConsensusReqReaderTest.java
@@ -41,6 +41,7 @@ import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.After;
import org.junit.Assert;
@@ -98,7 +99,10 @@ public class ConsensusReqReaderTest {
insertRowNode.setSearchIndex(1);
walNode.log(0, insertRowNode); // 1
insertTabletNode = getInsertTabletNode(devicePath, new long[] {2});
- walNode.log(0, insertTabletNode, 0, insertTabletNode.getRowCount()); // -1
+ walNode.log(
+ 0,
+ insertTabletNode,
+ Collections.singletonList(new Pair<>(0, insertTabletNode.getRowCount()))); // -1
walNode.rollWALFile();
// _1-1-1.wal
insertRowsNode = getInsertRowsNode(devicePath);
@@ -118,17 +122,32 @@ public class ConsensusReqReaderTest {
walNode.log(0, insertRowNode); // 3
insertTabletNode = getInsertTabletNode(devicePath, new long[] {4});
insertTabletNode.setSearchIndex(4);
- walNode.log(0, insertTabletNode, 0, insertTabletNode.getRowCount()); // 4
+ walNode.log(
+ 0,
+ insertTabletNode,
+ Collections.singletonList(new Pair<>(0, insertTabletNode.getRowCount()))); // 4
walNode.rollWALFile();
// _4-4-1.wal
- walNode.log(0, insertTabletNode, 0, insertTabletNode.getRowCount()); // 4
+ walNode.log(
+ 0,
+ insertTabletNode,
+ Collections.singletonList(new Pair<>(0, insertTabletNode.getRowCount()))); // 4
walNode.rollWALFile();
// _5-4-1.wal
- walNode.log(0, insertTabletNode, 0, insertTabletNode.getRowCount()); // 4
- walNode.log(0, insertTabletNode, 0, insertTabletNode.getRowCount()); // 4
+ walNode.log(
+ 0,
+ insertTabletNode,
+ Collections.singletonList(new Pair<>(0, insertTabletNode.getRowCount()))); // 4
+ walNode.log(
+ 0,
+ insertTabletNode,
+ Collections.singletonList(new Pair<>(0, insertTabletNode.getRowCount()))); // 4
insertTabletNode = getInsertTabletNode(devicePath, new long[] {5});
insertTabletNode.setSearchIndex(5);
- walNode.log(0, insertTabletNode, 0, insertTabletNode.getRowCount()); // 5
+ walNode.log(
+ 0,
+ insertTabletNode,
+ Collections.singletonList(new Pair<>(0, insertTabletNode.getRowCount()))); // 5
walNode.rollWALFile();
// _6-5-1.wal
insertRowNode = getInsertRowNode(devicePath);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNodeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNodeTest.java
index a1f1df61f42..2e1bc3945c8 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNodeTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNodeTest.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNod
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALFlushListener;
+import org.apache.tsfile.utils.Pair;
import org.junit.Test;
import java.util.ArrayList;
@@ -42,7 +43,11 @@ public class WALFakeNodeTest {
// log something
List<WALFlushListener> walFlushListeners = new ArrayList<>();
walFlushListeners.add(walNode.log(1, new InsertRowNode(new PlanNodeId("0"))));
- walFlushListeners.add(walNode.log(1, new InsertTabletNode(new PlanNodeId("1")), 0, 0));
+ walFlushListeners.add(
+ walNode.log(
+ 1,
+ new InsertTabletNode(new PlanNodeId("1")),
+ Collections.singletonList(new Pair<>(0, 0))));
walFlushListeners.add(
walNode.log(1, new DeleteDataNode(new PlanNodeId("2"), Collections.emptyList(), 0, 0)));
// check flush listeners
@@ -62,7 +67,11 @@ public class WALFakeNodeTest {
// log something
List<WALFlushListener> walFlushListeners = new ArrayList<>();
walFlushListeners.add(walNode.log(1, new InsertRowNode(new PlanNodeId("0"))));
- walFlushListeners.add(walNode.log(1, new InsertTabletNode(new PlanNodeId("1")), 0, 0));
+ walFlushListeners.add(
+ walNode.log(
+ 1,
+ new InsertTabletNode(new PlanNodeId("1")),
+ Collections.singletonList(new Pair<>(0, 0))));
walFlushListeners.add(
walNode.log(1, new DeleteDataNode(new PlanNodeId("2"), Collections.emptyList(), 0, 0)));
// check flush listeners
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java
index f72f55a3cb8..3282c228488 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java
@@ -45,6 +45,7 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.awaitility.Awaitility;
import org.junit.After;
@@ -169,7 +170,10 @@ public class WALNodeTest {
getInsertTabletNode(devicePath + memTableId, new long[] {i});
expectedInsertTabletNodes.add(insertTabletNode);
WALFlushListener walFlushListener =
- walNode.log(memTableId, insertTabletNode, 0, insertTabletNode.getRowCount());
+ walNode.log(
+ memTableId,
+ insertTabletNode,
+ Collections.singletonList(new Pair<>(0, insertTabletNode.getRowCount())));
walFlushListeners.add(walFlushListener);
}
}
@@ -297,7 +301,10 @@ public class WALNodeTest {
InsertTabletNode insertTabletNode =
getInsertTabletNode(devicePath + memTableId, new long[] {time});
WALFlushListener walFlushListener =
- walNode.log(memTableId, insertTabletNode, 0, insertTabletNode.getRowCount());
+ walNode.log(
+ memTableId,
+ insertTabletNode,
+ Collections.singletonList(new Pair<>(0, insertTabletNode.getRowCount())));
walFlushListeners.add(walFlushListener);
}
walNode.onMemTableFlushed(memTable);
(iotdb) 04/04: Merge branch 'master' of github.com:apache/iotdb into batch_wal_for_insertRelationalTablet
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch batch_wal_for_insertRelationalTablet
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5543d2507972f1ca99b15123329a15c4c5540850
Merge: d221dec6a87 502f64d15a0
Author: HTHou <hh...@outlook.com>
AuthorDate: Sat Oct 19 23:15:45 2024 +0800
Merge branch 'master' of github.com:apache/iotdb into batch_wal_for_insertRelationalTablet
.../db/it/IoTDBMultiIDsWithAttributesTableIT.java | 86 +++
.../it/query/recent/IoTDBGapFillTableIT.java | 40 +-
.../it/query/recent/IoTDBTableAggregationIT.java | 163 +++---
.../pipe/consensuspipe/ConsensusPipeManager.java | 4 +
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 21 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 4 +
.../db/consensus/DataRegionConsensusImpl.java | 2 +
.../dataregion/DataExecutionVisitor.java | 11 +-
.../agent/runtime/PipeDataNodeRuntimeAgent.java | 8 +-
.../agent/task/connection/PipeEventCollector.java | 41 +-
.../evolvable/batch/PipeTabletEventPlainBatch.java | 4 +
.../airgap/IoTDBDataNodeAirGapConnector.java | 44 --
.../airgap/IoTDBDataRegionAirGapConnector.java | 46 +-
.../airgap/IoTDBSchemaRegionAirGapConnector.java | 40 ++
.../pipeconsensus/PipeConsensusAsyncConnector.java | 12 +-
.../pipeconsensus/PipeConsensusSyncConnector.java | 80 ++-
.../PipeConsensusTabletBatchEventHandler.java | 2 +-
.../PipeConsensusTabletInsertNodeEventHandler.java | 2 +-
.../PipeConsensusTabletInsertionEventHandler.java | 2 +-
.../PipeConsensusTsFileInsertionEventHandler.java | 2 +-
.../request/PipeConsensusDeleteNodeReq.java | 126 +++++
.../async/IoTDBDataRegionAsyncConnector.java | 12 +-
.../thrift/sync/IoTDBDataNodeSyncConnector.java | 67 ---
.../thrift/sync/IoTDBDataRegionSyncConnector.java | 64 ++-
.../thrift/sync/IoTDBSchemaRegionConnector.java | 59 ++
.../consensus/ProgressIndexDataNodeManager.java | 52 +-
.../pipe/consensus/deletion/DeletionResource.java | 198 +++++++
.../deletion/DeletionResourceManager.java | 331 +++++++++++
.../deletion/persist/DeletionBuffer.java} | 25 +-
.../deletion/persist/PageCacheDeletionBuffer.java | 380 +++++++++++++
.../consensus/deletion/recover/DeletionReader.java | 101 ++++
.../PipeConsensusConnectorMetrics.java | 2 +-
.../{ => metric}/PipeConsensusReceiverMetrics.java | 2 +-
.../common/deletion/PipeDeleteDataNodeEvent.java | 186 +++++++
...ileInsertionEventTableParserTabletIterator.java | 12 +-
.../event/realtime/PipeRealtimeEventFactory.java | 4 +-
.../dataregion/DataRegionListeningFilter.java | 1 -
.../dataregion/IoTDBDataRegionExtractor.java | 4 +-
...ricalDataRegionTsFileAndDeletionExtractor.java} | 351 ++++++++----
.../realtime/PipeRealtimeDataRegionExtractor.java | 4 +
.../PipeRealtimeDataRegionHybridExtractor.java | 6 +-
.../PipeRealtimeDataRegionLogExtractor.java | 6 +-
.../PipeRealtimeDataRegionTsFileExtractor.java | 6 +-
.../realtime/assigner/PipeDataRegionAssigner.java | 20 +
.../listener/PipeInsertionDataNodeListener.java | 33 +-
.../matcher/CachedSchemaPatternMatcher.java | 4 +-
.../protocol/airgap/IoTDBAirGapReceiver.java | 8 +
.../pipeconsensus/PipeConsensusReceiver.java | 40 +-
.../process/gapfill/AbstractGapFillOperator.java | 2 +
.../relational/aggregation/AccumulatorFactory.java | 98 ++--
.../relational/aggregation/AvgAccumulator.java | 2 +-
.../relational/aggregation/CountAccumulator.java | 3 +-
.../relational/aggregation/ExtremeAccumulator.java | 306 ++++++++++
.../relational/aggregation/FirstAccumulator.java | 4 +-
.../relational/aggregation/FirstByAccumulator.java | 540 ++++++++++++++++++
.../relational/aggregation/LastAccumulator.java | 8 +-
.../relational/aggregation/LastByAccumulator.java | 495 ++++++++++++++++-
.../relational/aggregation/MaxAccumulator.java | 2 +-
.../relational/aggregation/MinAccumulator.java | 2 +-
.../relational/aggregation/SumAccumulator.java | 3 +-
.../aggregation/TableMaxByAccumulator.java | 73 +++
.../aggregation/TableMaxMinByBaseAccumulator.java | 486 ++++++++++++++++
.../aggregation/TableMinByAccumulator.java | 73 +++
.../grouped/GroupedExtremeAccumulator.java | 328 +++++++++++
.../grouped/GroupedFirstAccumulator.java | 425 ++++++++++++++
.../grouped/GroupedFirstByAccumulator.java | 544 ++++++++++++++++++
.../grouped/GroupedLastAccumulator.java | 425 ++++++++++++++
.../grouped/GroupedLastByAccumulator.java | 544 ++++++++++++++++++
.../aggregation/grouped/GroupedMaxAccumulator.java | 386 +++++++++++++
.../GroupedMaxByAccumulator.java} | 43 +-
.../grouped/GroupedMaxMinByBaseAccumulator.java | 614 +++++++++++++++++++++
.../aggregation/grouped/GroupedMinAccumulator.java | 386 +++++++++++++
.../GroupedMinByAccumulator.java} | 43 +-
.../GroupedSumAccumulator.java} | 123 ++---
.../aggregation/grouped/array/BinaryBigArray.java | 76 +++
.../aggregation/grouped/array/FloatBigArray.java | 165 ++++++
.../load/LoadTsFileToTableModelAnalyzer.java | 84 ++-
.../plan/planner/TableOperatorGenerator.java | 42 +-
.../ContinuousSameSearchIndexSeparatorNode.java | 11 +
.../planner/plan/node/write/DeleteDataNode.java | 188 +++++--
.../plan/planner/plan/node/write/InsertNode.java | 3 +-
.../plan/planner/plan/node/write/SearchNode.java | 5 +-
.../predicate/ConvertPredicateToFilterVisitor.java | 2 +-
.../ConvertPredicateToTimeFilterVisitor.java | 8 +-
.../relational/metadata/TableMetadataImpl.java | 22 +-
.../plan/relational/metadata/TableSchema.java | 34 +-
.../plan/relational/planner/QueryPlanner.java | 8 +-
.../plan/relational/planner/node/GapFillNode.java | 1 -
.../plan/relational/sql/parser/AstBuilder.java | 24 +-
.../plan/scheduler/load/LoadTsFileScheduler.java | 15 +-
.../iotdb/db/service/DataNodeShutdownHook.java | 4 +
.../db/storageengine/dataregion/DataRegion.java | 32 +-
.../dataregion/tsfile/TsFileResource.java | 10 +-
.../dataregion/wal/buffer/WALBuffer.java | 2 +-
.../db/storageengine/load/LoadTsFileManager.java | 21 +-
.../load/splitter/AlignedChunkData.java | 2 +-
.../db/storageengine/load/splitter/ChunkData.java | 4 +-
.../storageengine/load/splitter/DeletionData.java | 6 +-
.../load/splitter/NonAlignedChunkData.java | 2 +-
.../db/storageengine/load/splitter/TsFileData.java | 13 +-
.../{TsFileData.java => TsFileDataType.java} | 24 +-
.../iotdb/db/utils/constant/SqlConstant.java | 8 +-
.../db/pipe/consensus/DeletionRecoverTest.java | 85 +++
.../db/pipe/consensus/DeletionResourceTest.java | 211 +++++++
.../pipe/PipeEnrichedDeleteDataNodeSerdeTest.java | 4 +-
.../node/write/DeleteDataNodeSerdeTest.java | 4 +-
.../storageengine/dataregion/DataRegionTest.java | 195 +++++--
...eCrossSpaceCompactionWithFastPerformerTest.java | 55 +-
...sSpaceCompactionWithReadPointPerformerTest.java | 55 +-
.../InnerSeqCompactionWithFastPerformerTest.java | 18 +-
...nerSeqCompactionWithReadChunkPerformerTest.java | 18 +-
.../settle/SettleRequestHandlerTest.java | 10 +-
.../conf/iotdb-system.properties.template | 13 +
.../iotdb/commons/concurrent/ThreadName.java | 3 +
.../apache/iotdb/commons/conf/IoTDBConstant.java | 1 +
.../consensus/index/impl/SimpleProgressIndex.java | 38 +-
.../request/PipeConsensusRequestType.java | 1 +
.../pipe/datastructure/PersistentResource.java} | 27 +-
pom.xml | 2 +-
119 files changed, 9287 insertions(+), 945 deletions(-)
(iotdb) 01/04: try dev
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch batch_wal_for_insertRelationalTablet
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e8828b9dea3c778c0ac440bac78508dba1aef5ca
Author: HTHou <hh...@outlook.com>
AuthorDate: Fri Oct 18 22:41:58 2024 +0800
try dev
---
.../db/storageengine/dataregion/DataRegion.java | 158 ++++++++++-----------
.../dataregion/memtable/TsFileProcessor.java | 111 ++++++++-------
.../dataregion/wal/buffer/WALInfoEntry.java | 83 ++++++-----
.../dataregion/wal/node/IWALNode.java | 7 +-
.../dataregion/wal/node/WALFakeNode.java | 6 +-
.../storageengine/dataregion/wal/node/WALNode.java | 5 +-
.../dataregion/memtable/TsFileProcessorTest.java | 87 ++++++++++--
7 files changed, 276 insertions(+), 181 deletions(-)
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 0c39924e9e8..fba39e46471 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
@@ -1070,14 +1070,11 @@ public class DataRegion implements IDataRegionForQuery {
: Long.MAX_VALUE;
}
- private boolean splitAndInsert(
+ private void split(
InsertTabletNode insertTabletNode,
int loc,
int endOffset,
- TSStatus[] results,
- long[] costsForMetrics) {
- boolean noFailure = true;
-
+ Map<Long, Map<Boolean, List<Pair<Integer, Integer>>>> splitInfo) {
// before is first start point
int before = loc;
long beforeTime = insertTabletNode.getTimes()[before];
@@ -1086,7 +1083,6 @@ public class DataRegion implements IDataRegionForQuery {
// init flush time map
initFlushTimeMap(beforeTimePartition);
- int insertCnt = 0;
// if is sequence
boolean isSequence = false;
while (loc < endOffset) {
@@ -1098,27 +1094,7 @@ public class DataRegion implements IDataRegionForQuery {
if (timePartitionId != beforeTimePartition) {
initFlushTimeMap(timePartitionId);
lastFlushTime = getLastFlushTime(timePartitionId, insertTabletNode.getDeviceID(loc));
- // a new partition, insert the remaining of the previous partition
- noFailure =
- insertTabletToTsFileProcessor(
- insertTabletNode,
- before,
- loc,
- isSequence,
- results,
- beforeTimePartition,
- noFailure,
- costsForMetrics)
- && noFailure;
- if (before < loc) {
- insertCnt += 1;
- logger.debug(
- "insertTabletToTsFileProcessor, insertCnt:{}, noFailure:{}, before:{}, loc:{}",
- insertCnt,
- noFailure,
- before,
- loc);
- }
+ updateSplitInfo(splitInfo, beforeTimePartition, isSequence, new Pair<>(before, loc));
before = loc;
beforeTimePartition = timePartitionId;
isSequence = time > lastFlushTime;
@@ -1127,27 +1103,8 @@ public class DataRegion implements IDataRegionForQuery {
if (time > lastFlushTime) {
// the same partition and switch to sequence data
// insert previous range into unsequence
- noFailure =
- insertTabletToTsFileProcessor(
- insertTabletNode,
- before,
- loc,
- isSequence,
- results,
- beforeTimePartition,
- noFailure,
- costsForMetrics)
- && noFailure;
+ updateSplitInfo(splitInfo, beforeTimePartition, isSequence, new Pair<>(before, loc));
before = loc;
- if (before < loc) {
- insertCnt += 1;
- logger.debug(
- "insertTabletToTsFileProcessor, insertCnt:{}, noFailure:{}, before:{}, loc:{}",
- insertCnt,
- noFailure,
- before,
- loc);
- }
isSequence = true;
}
}
@@ -1157,26 +1114,56 @@ public class DataRegion implements IDataRegionForQuery {
// do not forget last part
if (before < loc) {
- noFailure =
- insertTabletToTsFileProcessor(
- insertTabletNode,
- before,
- loc,
- isSequence,
- results,
- beforeTimePartition,
- noFailure,
- costsForMetrics)
- && noFailure;
- insertCnt += 1;
- logger.debug(
- "insertTabletToTsFileProcessor, insertCnt:{}, noFailure:{}, before:{}, loc:{}",
- insertCnt,
- noFailure,
- before,
- loc);
+ updateSplitInfo(splitInfo, beforeTimePartition, isSequence, new Pair<>(before, loc));
+ }
+ }
+
+ private void updateSplitInfo(
+ Map<Long, Map<Boolean, List<Pair<Integer, Integer>>>> splitInfo,
+ long partitionId,
+ boolean isSequence,
+ Pair<Integer, Integer> newRange) {
+ List<Pair<Integer, Integer>> rangeList =
+ splitInfo
+ .computeIfAbsent(partitionId, k -> new HashMap<>())
+ .computeIfAbsent(isSequence, k -> new ArrayList<>());
+
+ if (!rangeList.isEmpty()) {
+ Pair<Integer, Integer> lastRange = rangeList.get(rangeList.size() - 1);
+ if (lastRange.right.equals(newRange.left)) {
+ lastRange.right = newRange.right;
+ return;
+ }
+ }
+ if (newRange.left < newRange.right) {
+ rangeList.add(newRange);
}
+ }
+ private boolean insert(
+ InsertTabletNode insertTabletNode,
+ Map<Long, Map<Boolean, List<Pair<Integer, Integer>>>> splitMap,
+ TSStatus[] results,
+ long[] costsForMetrics) {
+ boolean noFailure = true;
+ for (Entry<Long, Map<Boolean, List<Pair<Integer, Integer>>>> entry : splitMap.entrySet()) {
+ long timePartitionId = entry.getKey();
+ Map<Boolean, List<Pair<Integer, Integer>>> splitInfo = entry.getValue();
+ for (Entry<Boolean, List<Pair<Integer, Integer>>> splitEntry : splitInfo.entrySet()) {
+ boolean isSequence = splitEntry.getKey();
+ List<Pair<Integer, Integer>> rangeList = splitEntry.getValue();
+ noFailure =
+ insertTabletToTsFileProcessor(
+ insertTabletNode,
+ rangeList,
+ isSequence,
+ results,
+ timePartitionId,
+ noFailure,
+ costsForMetrics)
+ && noFailure;
+ }
+ }
return noFailure;
}
@@ -1225,12 +1212,14 @@ public class DataRegion implements IDataRegionForQuery {
List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs =
insertTabletNode.splitByDevice(loc, insertTabletNode.getRowCount());
int start = loc;
+ Map<Long, Map<Boolean, List<Pair<Integer, Integer>>>> splitInfo = new HashMap<>();
for (Pair<IDeviceID, Integer> deviceEndOffsetPair : deviceEndOffsetPairs) {
int end = deviceEndOffsetPair.getRight();
- noFailure =
- noFailure && splitAndInsert(insertTabletNode, start, end, results, costsForMetrics);
+ split(insertTabletNode, start, end, splitInfo);
start = end;
}
+ noFailure = noFailure && insert(insertTabletNode, splitInfo, results, costsForMetrics);
+
if (CommonDescriptor.getInstance().getConfig().isLastCacheEnable()
&& !insertTabletNode.isGeneratedByRemoteConsensusLeader()) {
// disable updating last cache on follower
@@ -1307,39 +1296,50 @@ public class DataRegion implements IDataRegionForQuery {
*
* @param insertTabletNode insert a tablet of a device
* @param sequence whether is sequence
- * @param start start index of rows to be inserted in insertTabletPlan
- * @param end end index of rows to be inserted in insertTabletPlan
+ * @param rangeList start and end index list of rows to be inserted in insertTabletPlan
* @param results result array
* @param timePartitionId time partition id
* @return false if any failure occurs when inserting the tablet, true otherwise
*/
private boolean insertTabletToTsFileProcessor(
InsertTabletNode insertTabletNode,
- int start,
- int end,
+ List<Pair<Integer, Integer>> rangeList,
boolean sequence,
TSStatus[] results,
long timePartitionId,
boolean noFailure,
long[] costsForMetrics) {
// return when start >= end or all measurement failed
- if (start >= end || insertTabletNode.allMeasurementFailed()) {
+ // if (start >= end || insertTabletNode.allMeasurementFailed()) {
+ // if (logger.isDebugEnabled()) {
+ // logger.debug(
+ // "Won't insert tablet {}, because {}",
+ // insertTabletNode.getSearchIndex(),
+ // start >= end ? "start >= end" : "insertTabletNode allMeasurementFailed");
+ // }
+ // return true;
+ // }
+ if (insertTabletNode.allMeasurementFailed()) {
if (logger.isDebugEnabled()) {
logger.debug(
"Won't insert tablet {}, because {}",
insertTabletNode.getSearchIndex(),
- start >= end ? "start >= end" : "insertTabletNode allMeasurementFailed");
+ "insertTabletNode allMeasurementFailed");
}
return true;
}
TsFileProcessor tsFileProcessor = getOrCreateTsFileProcessor(timePartitionId, sequence);
if (tsFileProcessor == null) {
- for (int i = start; i < end; i++) {
- results[i] =
- RpcUtils.getStatus(
- TSStatusCode.INTERNAL_SERVER_ERROR,
- "can not create TsFileProcessor, timePartitionId: " + timePartitionId);
+ for (Pair<Integer, Integer> rangePair : rangeList) {
+ int start = rangePair.getLeft();
+ int end = rangePair.getRight();
+ for (int i = start; i < end; i++) {
+ results[i] =
+ RpcUtils.getStatus(
+ TSStatusCode.INTERNAL_SERVER_ERROR,
+ "can not create TsFileProcessor, timePartitionId: " + timePartitionId);
+ }
}
return false;
}
@@ -1349,7 +1349,7 @@ public class DataRegion implements IDataRegionForQuery {
try {
tsFileProcessor.insertTablet(
- insertTabletNode, start, end, results, noFailure, costsForMetrics);
+ insertTabletNode, rangeList, results, noFailure, costsForMetrics);
} catch (WriteProcessRejectException e) {
logger.warn("insert to TsFileProcessor rejected, {}", e.getMessage());
return false;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index f165905484b..52462955e52 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -451,26 +451,32 @@ public class TsFileProcessor {
private long[] scheduleMemoryBlock(
InsertTabletNode insertTabletNode,
- int start,
- int end,
+ List<Pair<Integer, Integer>> rangeList,
TSStatus[] results,
boolean noFailure,
long[] costsForMetrics)
throws WriteProcessException {
- long[] memIncrements;
- try {
- long memControlStartTime = System.nanoTime();
- memIncrements = checkMemCost(insertTabletNode, start, end, noFailure, results);
- // recordScheduleMemoryBlockCost
- costsForMetrics[1] += System.nanoTime() - memControlStartTime;
- } catch (WriteProcessException e) {
- for (int i = start; i < end; i++) {
- results[i] = RpcUtils.getStatus(TSStatusCode.WRITE_PROCESS_REJECT, e.getMessage());
+ long memControlStartTime = System.nanoTime();
+ long[] totalMemIncrements = new long[NUM_MEM_TO_ESTIMATE];
+ for (Pair<Integer, Integer> range : rangeList) {
+ int start = range.left;
+ int end = range.right;
+ try {
+ long[] memIncrements = checkMemCost(insertTabletNode, start, end, noFailure, results);
+ for (int i = 0; i < memIncrements.length; i++) {
+ totalMemIncrements[i] += memIncrements[i];
+ }
+ } catch (WriteProcessException e) {
+ for (int i = start; i < end; i++) {
+ results[i] = RpcUtils.getStatus(TSStatusCode.WRITE_PROCESS_REJECT, e.getMessage());
+ }
+ throw new WriteProcessException(e);
}
- throw new WriteProcessException(e);
}
+ // recordScheduleMemoryBlockCost
+ costsForMetrics[1] += System.nanoTime() - memControlStartTime;
- return memIncrements;
+ return totalMemIncrements;
}
private long[] checkMemCost(
@@ -526,14 +532,12 @@ public class TsFileProcessor {
* non-null value, e.g., {1, null, 3, null, 5} will be {1, 3, 5, null, 5}
*
* @param insertTabletNode insert a tablet of a device
- * @param start start index of rows to be inserted in insertTabletPlan
- * @param end end index of rows to be inserted in insertTabletPlan
+ * @param rangeList start and end index list of rows to be inserted in insertTabletPlan
* @param results result array
*/
public void insertTablet(
InsertTabletNode insertTabletNode,
- int start,
- int end,
+ List<Pair<Integer, Integer>> rangeList,
TSStatus[] results,
boolean noFailure,
long[] costsForMetrics)
@@ -542,18 +546,22 @@ public class TsFileProcessor {
ensureMemTable(costsForMetrics);
long[] memIncrements =
- scheduleMemoryBlock(insertTabletNode, start, end, results, noFailure, costsForMetrics);
+ scheduleMemoryBlock(insertTabletNode, rangeList, results, noFailure, costsForMetrics);
long startTime = System.nanoTime();
WALFlushListener walFlushListener;
try {
- walFlushListener = walNode.log(workMemTable.getMemTableId(), insertTabletNode, start, end);
+ walFlushListener = walNode.log(workMemTable.getMemTableId(), insertTabletNode, rangeList);
if (walFlushListener.waitForResult() == WALFlushListener.Status.FAILURE) {
throw walFlushListener.getCause();
}
} catch (Exception e) {
- for (int i = start; i < end; i++) {
- results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage());
+ for (Pair<Integer, Integer> rangePair : rangeList) {
+ int start = rangePair.getLeft();
+ int end = rangePair.getRight();
+ for (int i = start; i < end; i++) {
+ results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage());
+ }
}
rollbackMemoryInfo(memIncrements);
throw new WriteProcessException(e);
@@ -576,42 +584,47 @@ public class TsFileProcessor {
insertTabletNode,
tsFileResource);
- try {
- if (insertTabletNode.isAligned()) {
- workMemTable.insertAlignedTablet(insertTabletNode, start, end, noFailure ? null : results);
- } else {
- workMemTable.insertTablet(insertTabletNode, start, end);
+ for (Pair<Integer, Integer> rangePair : rangeList) {
+ int start = rangePair.getLeft();
+ int end = rangePair.getRight();
+ try {
+ if (insertTabletNode.isAligned()) {
+ workMemTable.insertAlignedTablet(
+ insertTabletNode, start, end, noFailure ? null : results);
+ } else {
+ workMemTable.insertTablet(insertTabletNode, start, end);
+ }
+ } catch (WriteProcessException e) {
+ for (int i = start; i < end; i++) {
+ results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage());
+ }
+ throw new WriteProcessException(e);
}
- } catch (WriteProcessException e) {
for (int i = start; i < end; i++) {
- results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage());
+ results[i] = RpcUtils.SUCCESS_STATUS;
}
- throw new WriteProcessException(e);
- }
- for (int i = start; i < end; i++) {
- results[i] = RpcUtils.SUCCESS_STATUS;
- }
- final List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs =
- insertTabletNode.splitByDevice(start, end);
- tsFileResource.updateStartTime(
- deviceEndOffsetPairs.get(0).left, insertTabletNode.getTimes()[start]);
- if (!sequence) {
- // For sequence tsfile, we update the endTime only when the file is prepared to be closed.
- // For unsequence tsfile, we have to update the endTime for each insertion.
- tsFileResource.updateEndTime(
- deviceEndOffsetPairs.get(0).left,
- insertTabletNode.getTimes()[deviceEndOffsetPairs.get(0).right - 1]);
- }
- for (int i = 1; i < deviceEndOffsetPairs.size(); i++) {
- // the end offset of i - 1 is the start offset of i
+ final List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs =
+ insertTabletNode.splitByDevice(start, end);
tsFileResource.updateStartTime(
- deviceEndOffsetPairs.get(i).left,
- insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i - 1).right]);
+ deviceEndOffsetPairs.get(0).left, insertTabletNode.getTimes()[start]);
if (!sequence) {
+ // For sequence tsfile, we update the endTime only when the file is prepared to be closed.
+ // For unsequence tsfile, we have to update the endTime for each insertion.
tsFileResource.updateEndTime(
+ deviceEndOffsetPairs.get(0).left,
+ insertTabletNode.getTimes()[deviceEndOffsetPairs.get(0).right - 1]);
+ }
+ for (int i = 1; i < deviceEndOffsetPairs.size(); i++) {
+ // the end offset of i - 1 is the start offset of i
+ tsFileResource.updateStartTime(
deviceEndOffsetPairs.get(i).left,
- insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i).right - 1]);
+ insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i - 1).right]);
+ if (!sequence) {
+ tsFileResource.updateEndTime(
+ deviceEndOffsetPairs.get(i).left,
+ insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i).right - 1]);
+ }
}
}
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 8b327b4880b..08239c1f329 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
@@ -19,11 +19,16 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.buffer;
+import java.util.ArrayList;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode;
+import org.apache.tsfile.utils.Pair;
+
+import java.util.Collections;
+import java.util.List;
import java.util.Objects;
/** This entry class stores info for persistence. */
@@ -42,19 +47,24 @@ public class WALInfoEntry extends WALEntry {
public WALInfoEntry(long memTableId, WALEntryValue value) {
this(memTableId, value, config.getWalMode() == WALMode.SYNC);
if (value instanceof InsertTabletNode) {
- tabletInfo = new TabletInfo(0, ((InsertTabletNode) value).getRowCount());
+ tabletInfo =
+ new TabletInfo(
+ Collections.singletonList(new Pair<>(0, ((InsertTabletNode) value).getRowCount())));
}
}
- public WALInfoEntry(long memTableId, InsertTabletNode value, int tabletStart, int tabletEnd) {
+ public WALInfoEntry(
+ long memTableId, InsertTabletNode value, List<Pair<Integer, Integer>> tabletRangeList) {
this(memTableId, value, config.getWalMode() == WALMode.SYNC);
- tabletInfo = new TabletInfo(tabletStart, tabletEnd);
+ tabletInfo = new TabletInfo(tabletRangeList);
}
WALInfoEntry(WALEntryType type, long memTableId, WALEntryValue value) {
super(type, memTableId, value, false);
if (value instanceof InsertTabletNode) {
- tabletInfo = new TabletInfo(0, ((InsertTabletNode) value).getRowCount());
+ tabletInfo =
+ new TabletInfo(
+ Collections.singletonList(new Pair<>(0, ((InsertTabletNode) value).getRowCount())));
}
}
@@ -65,18 +75,21 @@ public class WALInfoEntry extends WALEntry {
@Override
public void serialize(IWALByteBufferView buffer) {
- buffer.put(type.getCode());
- buffer.putLong(memTableId);
switch (type) {
case INSERT_TABLET_NODE:
- ((InsertTabletNode) value)
- .serializeToWAL(buffer, tabletInfo.tabletStart, tabletInfo.tabletEnd);
+ for (Pair<Integer, Integer> range : tabletInfo.tabletRangeList) {
+ buffer.put(type.getCode());
+ buffer.putLong(memTableId);
+ ((InsertTabletNode) value).serializeToWAL(buffer, range.left, range.right);
+ }
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:
@@ -87,35 +100,35 @@ public class WALInfoEntry extends WALEntry {
}
private static class TabletInfo {
- // start row of insert tablet
- private final int tabletStart;
- // end row of insert tablet
- private final int tabletEnd;
-
- public TabletInfo(int tabletStart, int tabletEnd) {
- this.tabletStart = tabletStart;
- this.tabletEnd = tabletEnd;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(tabletStart, tabletEnd);
+ // // start row of insert tablet
+ // private final int tabletStart;
+ // // end row of insert tablet
+ // private final int tabletEnd;
+ List<Pair<Integer, Integer>> tabletRangeList;
+
+ public TabletInfo(List<Pair<Integer, Integer>> tabletRangeList) {
+ this.tabletRangeList = new ArrayList<>(tabletRangeList);
}
- @Override
- public boolean equals(Object obj) {
- if (obj == this) {
- return true;
- }
- if (obj == null) {
- return false;
- }
- if (!(obj instanceof TabletInfo)) {
- return false;
- }
- TabletInfo other = (TabletInfo) obj;
- return this.tabletStart == other.tabletStart && this.tabletEnd == other.tabletEnd;
- }
+ // @Override
+ // public int hashCode() {
+ // return Objects.hash(tabletStart, tabletEnd);
+ // }
+ //
+ // @Override
+ // public boolean equals(Object obj) {
+ // if (obj == this) {
+ // return true;
+ // }
+ // if (obj == null) {
+ // return false;
+ // }
+ // if (!(obj instanceof TabletInfo)) {
+ // return false;
+ // }
+ // TabletInfo other = (TabletInfo) obj;
+ // return this.tabletStart == other.tabletStart && this.tabletEnd == other.tabletEnd;
+ // }
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/IWALNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/IWALNode.java
index 2baf45f3ae5..767d557744c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/IWALNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/IWALNode.java
@@ -30,6 +30,10 @@ import org.apache.iotdb.db.storageengine.dataregion.flush.FlushListener;
import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALFlushListener;
+import org.apache.tsfile.utils.Pair;
+
+import java.util.List;
+
/** This interface provides uniform interface for writing wal and making checkpoints. */
public interface IWALNode extends FlushListener, AutoCloseable, ConsensusReqReader, DataSet {
@@ -40,7 +44,8 @@ public interface IWALNode extends FlushListener, AutoCloseable, ConsensusReqRead
WALFlushListener log(long memTableId, InsertRowsNode insertRowsNode);
/** Log InsertTabletNode. */
- WALFlushListener log(long memTableId, InsertTabletNode insertTabletNode, int start, int end);
+ WALFlushListener log(
+ long memTableId, InsertTabletNode insertTabletNode, List<Pair<Integer, Integer>> rangeList);
/** Log DeleteDataNode. */
WALFlushListener log(long memTableId, DeleteDataNode deleteDataNode);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNode.java
index 38b69f1162b..0aee5f56bf0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNode.java
@@ -29,6 +29,10 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALException;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.AbstractResultListener.Status;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALFlushListener;
+import org.apache.tsfile.utils.Pair;
+
+import java.util.List;
+
/** This class provides fake wal node when wal is disabled or exception happens. */
public class WALFakeNode implements IWALNode {
private final Status status;
@@ -59,7 +63,7 @@ public class WALFakeNode implements IWALNode {
@Override
public WALFlushListener log(
- long memTableId, InsertTabletNode insertTabletNode, int start, int end) {
+ long memTableId, InsertTabletNode insertTabletNode, List<Pair<Integer, Integer>> rangeList) {
return getResult();
}
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 607a99a5673..b0d165545a0 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
@@ -59,6 +59,7 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALFlushL
import org.apache.commons.lang3.StringUtils;
import org.apache.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.utils.TsFileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -151,12 +152,12 @@ public class WALNode implements IWALNode {
@Override
public WALFlushListener log(
- long memTableId, InsertTabletNode insertTabletNode, int start, int end) {
+ long memTableId, InsertTabletNode insertTabletNode, List<Pair<Integer, Integer>> rangeList) {
logger.debug(
"WAL node-{} logs insertTabletNode, the search index is {}.",
identifier,
insertTabletNode.getSearchIndex());
- WALEntry walEntry = new WALInfoEntry(memTableId, insertTabletNode, start, end);
+ WALEntry walEntry = new WALInfoEntry(memTableId, insertTabletNode, rangeList);
return log(walEntry);
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
index 5a7e260115b..775fef2ba4d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
@@ -47,6 +47,7 @@ import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.read.reader.IPointReader;
+import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.write.record.TSRecord;
import org.apache.tsfile.write.record.datapoint.DataPoint;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -303,14 +304,27 @@ public class TsFileProcessorTest {
this.sgInfo.initTsFileProcessorInfo(processor);
SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, processor);
// Test Tablet
- processor.insertTablet(genInsertTableNode(0, true), 0, 10, new TSStatus[10], true, new long[4]);
+ processor.insertTablet(
+ genInsertTableNode(0, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
IMemTable memTable = processor.getWorkMemTable();
Assert.assertEquals(1596808, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNode(100, true), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNode(100, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(1596808, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNode(200, true), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNode(200, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(1596808, memTable.getTVListsRamCost());
Assert.assertEquals(90000, memTable.getTotalPointsNum());
Assert.assertEquals(720360, memTable.memSize());
@@ -341,29 +355,62 @@ public class TsFileProcessorTest {
this.sgInfo.initTsFileProcessorInfo(processor);
SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, processor);
// Test Tablet
- processor.insertTablet(genInsertTableNode(0, true), 0, 10, new TSStatus[10], true, new long[4]);
+ processor.insertTablet(
+ genInsertTableNode(0, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
IMemTable memTable = processor.getWorkMemTable();
Assert.assertEquals(1596808, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNodeFors3000ToS6000(0, true), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNodeFors3000ToS6000(0, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(3192808, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNode(100, true), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNode(100, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(3192808, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNodeFors3000ToS6000(100, true), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNodeFors3000ToS6000(100, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(3192808, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNode(200, true), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNode(200, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(3192808, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNodeFors3000ToS6000(200, true), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNodeFors3000ToS6000(200, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(3192808, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNode(300, true), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNode(300, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(6385616, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNodeFors3000ToS6000(300, true), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNodeFors3000ToS6000(300, true),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(6385616, memTable.getTVListsRamCost());
Assert.assertEquals(240000, memTable.getTotalPointsNum());
@@ -403,14 +450,26 @@ public class TsFileProcessorTest {
SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, processor);
// Test tablet
processor.insertTablet(
- genInsertTableNode(0, false), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNode(0, false),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
IMemTable memTable = processor.getWorkMemTable();
Assert.assertEquals(3192000, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNode(100, false), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNode(100, false),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(3192000, memTable.getTVListsRamCost());
processor.insertTablet(
- genInsertTableNode(200, false), 0, 10, new TSStatus[10], true, new long[4]);
+ genInsertTableNode(200, false),
+ Collections.singletonList(new Pair<>(0, 10)),
+ new TSStatus[10],
+ true,
+ new long[4]);
Assert.assertEquals(3192000, memTable.getTVListsRamCost());
Assert.assertEquals(90000, memTable.getTotalPointsNum());
Assert.assertEquals(1440000, memTable.memSize());