Skip to content

Commit 6fd3870

Browse files
CaideyipiCaideyipi
authored
Pipe: Optimized the table model writing latency by batching & Fixed the NPE caused by tablet event sorting
1 parent 1165497 commit 6fd3870

File tree

3 files changed

+137
-43
lines changed

3 files changed

+137
-43
lines changed

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventPlainBatch.java‎

Lines changed: 90 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -22,14 +22,17 @@
2222
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
2323
import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
2424
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
25+
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
2526
import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletBatchReqV2;
2627
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
27-
import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
28+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode;
2829
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
2930

3031
import org.apache.tsfile.utils.Pair;
3132
import org.apache.tsfile.utils.PublicBAOS;
33+
import org.apache.tsfile.utils.RamUsageEstimator;
3234
import org.apache.tsfile.utils.ReadWriteIOUtils;
35+
import org.apache.tsfile.write.record.Tablet;
3336

3437
import java.io.DataOutputStream;
3538
import java.io.IOException;
@@ -38,7 +41,7 @@
3841
import java.util.HashMap;
3942
import java.util.List;
4043
import java.util.Map;
41-
import java.util.Objects;
44+
import java.util.concurrent.atomic.AtomicLong;
4245

4346
public class PipeTabletEventPlainBatch extends PipeTabletEventBatch {
4447

@@ -51,13 +54,13 @@ public class PipeTabletEventPlainBatch extends PipeTabletEventBatch {
5154
private final List<String> insertNodeDataBases = new ArrayList<>();
5255
private final List<String> tabletDataBases = new ArrayList<>();
5356

57+
// database -> tableName -> Pair<size, tablets to batch>
58+
private final Map<String, Map<String, Pair<Integer, List<Tablet>>>> tableModelTabletMap =
59+
new HashMap<>();
60+
5461
// Used to rate limit when transferring data
5562
private final Map<Pair<String, Long>, Long> pipe2BytesAccumulated = new HashMap<>();
5663

57-
PipeTabletEventPlainBatch(final int maxDelayInMs, final long requestMaxBatchSizeInBytes) {
58-
super(maxDelayInMs, requestMaxBatchSizeInBytes, null);
59-
}
60-
6164
PipeTabletEventPlainBatch(
6265
final int maxDelayInMs,
6366
final long requestMaxBatchSizeInBytes,
@@ -66,9 +69,8 @@ public class PipeTabletEventPlainBatch extends PipeTabletEventBatch {
6669
}
6770

6871
@Override
69-
protected boolean constructBatch(final TabletInsertionEvent event)
70-
throws WALPipeException, IOException {
71-
final int bufferSize = buildTabletInsertionBuffer(event);
72+
protected boolean constructBatch(final TabletInsertionEvent event) throws IOException {
73+
final long bufferSize = buildTabletInsertionBuffer(event);
7274
totalBufferSize += bufferSize;
7375
pipe2BytesAccumulated.compute(
7476
new Pair<>(
@@ -89,11 +91,45 @@ public synchronized void onSuccess() {
8991
binaryDataBases.clear();
9092
insertNodeDataBases.clear();
9193
tabletDataBases.clear();
94+
tableModelTabletMap.clear();
9295

9396
pipe2BytesAccumulated.clear();
9497
}
9598

9699
public PipeTransferTabletBatchReqV2 toTPipeTransferReq() throws IOException {
100+
for (final Map.Entry<String, Map<String, Pair<Integer, List<Tablet>>>> insertTablets :
101+
tableModelTabletMap.entrySet()) {
102+
final String databaseName = insertTablets.getKey();
103+
for (final Map.Entry<String, Pair<Integer, List<Tablet>>> tabletEntry :
104+
insertTablets.getValue().entrySet()) {
105+
final List<Tablet> batchTablets = new ArrayList<>();
106+
for (final Tablet tablet : tabletEntry.getValue().getRight()) {
107+
boolean success = false;
108+
for (final Tablet batchTablet : batchTablets) {
109+
if (batchTablet.append(tablet, tabletEntry.getValue().getLeft())) {
110+
success = true;
111+
break;
112+
}
113+
}
114+
if (!success) {
115+
batchTablets.add(tablet);
116+
}
117+
}
118+
for (final Tablet batchTablet : batchTablets) {
119+
try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
120+
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
121+
batchTablet.serialize(outputStream);
122+
ReadWriteIOUtils.write(true, outputStream);
123+
tabletBuffers.add(
124+
ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()));
125+
}
126+
tabletDataBases.add(databaseName);
127+
}
128+
}
129+
}
130+
131+
tableModelTabletMap.clear();
132+
97133
return PipeTransferTabletBatchReqV2.toTPipeTransferReq(
98134
binaryBuffers,
99135
insertNodeBuffers,
@@ -111,57 +147,71 @@ public Map<Pair<String, Long>, Long> getPipe2BytesAccumulated() {
111147
return pipe2BytesAccumulated;
112148
}
113149

114-
private int buildTabletInsertionBuffer(final TabletInsertionEvent event)
115-
throws IOException, WALPipeException {
116-
int databaseEstimateSize = 0;
150+
private long buildTabletInsertionBuffer(final TabletInsertionEvent event) throws IOException {
151+
long estimateSize = 0;
117152
final ByteBuffer buffer;
118153
if (event instanceof PipeInsertNodeTabletInsertionEvent) {
119154
final PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletInsertionEvent =
120155
(PipeInsertNodeTabletInsertionEvent) event;
121-
// Read the bytebuffer from the wal file and transfer it directly without serializing or
122-
// deserializing if possible
123156
final InsertNode insertNode = pipeInsertNodeTabletInsertionEvent.getInsertNode();
124-
if (Objects.isNull(insertNode)) {
125-
buffer = pipeInsertNodeTabletInsertionEvent.getByteBuffer();
126-
binaryBuffers.add(buffer);
127-
if (pipeInsertNodeTabletInsertionEvent.isTableModelEvent()) {
128-
databaseEstimateSize =
129-
pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName().length();
130-
binaryDataBases.add(pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName());
131-
} else {
132-
databaseEstimateSize = 4;
133-
binaryDataBases.add(TREE_MODEL_DATABASE_PLACEHOLDER);
134-
}
135-
} else {
157+
if (!(insertNode instanceof RelationalInsertTabletNode)) {
136158
buffer = insertNode.serializeToByteBuffer();
137159
insertNodeBuffers.add(buffer);
138160
if (pipeInsertNodeTabletInsertionEvent.isTableModelEvent()) {
139-
databaseEstimateSize =
140-
pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName().length();
161+
estimateSize =
162+
RamUsageEstimator.sizeOf(
163+
pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName());
141164
insertNodeDataBases.add(pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName());
142165
} else {
143-
databaseEstimateSize = 4;
166+
estimateSize = 4;
144167
insertNodeDataBases.add(TREE_MODEL_DATABASE_PLACEHOLDER);
145168
}
169+
estimateSize += buffer.limit();
170+
} else {
171+
for (final Tablet tablet :
172+
((PipeInsertNodeTabletInsertionEvent) event).convertToTablets()) {
173+
estimateSize +=
174+
constructTabletBatch(
175+
tablet, pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName());
176+
}
146177
}
147178
} else {
148179
final PipeRawTabletInsertionEvent pipeRawTabletInsertionEvent =
149180
(PipeRawTabletInsertionEvent) event;
150-
try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
151-
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
152-
pipeRawTabletInsertionEvent.convertToTablet().serialize(outputStream);
153-
ReadWriteIOUtils.write(pipeRawTabletInsertionEvent.isAligned(), outputStream);
154-
buffer = ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
155-
}
156-
tabletBuffers.add(buffer);
157181
if (pipeRawTabletInsertionEvent.isTableModelEvent()) {
158-
databaseEstimateSize = pipeRawTabletInsertionEvent.getTableModelDatabaseName().length();
159-
tabletDataBases.add(pipeRawTabletInsertionEvent.getTableModelDatabaseName());
182+
estimateSize =
183+
constructTabletBatch(
184+
pipeRawTabletInsertionEvent.convertToTablet(),
185+
pipeRawTabletInsertionEvent.getTableModelDatabaseName());
160186
} else {
161-
databaseEstimateSize = 4;
187+
try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
188+
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
189+
pipeRawTabletInsertionEvent.convertToTablet().serialize(outputStream);
190+
ReadWriteIOUtils.write(pipeRawTabletInsertionEvent.isAligned(), outputStream);
191+
buffer = ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
192+
}
193+
estimateSize = 4 + buffer.limit();
194+
tabletBuffers.add(buffer);
162195
tabletDataBases.add(TREE_MODEL_DATABASE_PLACEHOLDER);
163196
}
164197
}
165-
return buffer.limit() + databaseEstimateSize;
198+
199+
return estimateSize;
200+
}
201+
202+
private long constructTabletBatch(final Tablet tablet, final String databaseName) {
203+
final AtomicLong size = new AtomicLong(0);
204+
final Pair<Integer, List<Tablet>> currentBatch =
205+
tableModelTabletMap
206+
.computeIfAbsent(
207+
databaseName,
208+
k -> {
209+
size.addAndGet(RamUsageEstimator.sizeOf(databaseName));
210+
return new HashMap<>();
211+
})
212+
.computeIfAbsent(tablet.getTableName(), k -> new Pair<>(0, new ArrayList<>()));
213+
currentBatch.setLeft(currentBatch.getLeft() + tablet.getRowSize());
214+
currentBatch.getRight().add(tablet);
215+
return PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet) + 4;
166216
}
167217
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2.java‎

Lines changed: 46 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,9 @@
3939
import java.io.IOException;
4040
import java.nio.ByteBuffer;
4141
import java.util.ArrayList;
42+
import java.util.HashMap;
4243
import java.util.List;
44+
import java.util.Map;
4345
import java.util.Objects;
4446

4547
public class PipeTransferTabletBatchReqV2 extends TPipeTransferReq {
@@ -62,14 +64,31 @@ public List<InsertBaseStatement> constructStatements() {
6264

6365
final List<InsertRowStatement> insertRowStatementList = new ArrayList<>();
6466
final List<InsertTabletStatement> insertTabletStatementList = new ArrayList<>();
67+
final Map<String, List<InsertRowStatement>> tableModelDatabaseInsertRowStatementMap =
68+
new HashMap<>();
6569

6670
for (final PipeTransferTabletBinaryReqV2 binaryReq : binaryReqs) {
6771
final InsertBaseStatement statement = binaryReq.constructStatement();
6872
if (statement.isEmpty()) {
6973
continue;
7074
}
7175
if (statement.isWriteToTable()) {
72-
statements.add(statement);
76+
if (statement instanceof InsertRowStatement) {
77+
tableModelDatabaseInsertRowStatementMap
78+
.computeIfAbsent(statement.getDatabaseName().get(), k -> new ArrayList<>())
79+
.add((InsertRowStatement) statement);
80+
} else if (statement instanceof InsertTabletStatement) {
81+
statements.add(statement);
82+
} else if (statement instanceof InsertRowsStatement) {
83+
tableModelDatabaseInsertRowStatementMap
84+
.computeIfAbsent(statement.getDatabaseName().get(), k -> new ArrayList<>())
85+
.addAll(((InsertRowsStatement) statement).getInsertRowStatementList());
86+
} else {
87+
throw new UnsupportedOperationException(
88+
String.format(
89+
"unknown InsertBaseStatement %s constructed from PipeTransferTabletBinaryReqV2.",
90+
binaryReq));
91+
}
7392
continue;
7493
}
7594
if (statement instanceof InsertRowStatement) {
@@ -93,7 +112,22 @@ public List<InsertBaseStatement> constructStatements() {
93112
continue;
94113
}
95114
if (statement.isWriteToTable()) {
96-
statements.add(statement);
115+
if (statement instanceof InsertRowStatement) {
116+
tableModelDatabaseInsertRowStatementMap
117+
.computeIfAbsent(statement.getDatabaseName().get(), k -> new ArrayList<>())
118+
.add((InsertRowStatement) statement);
119+
} else if (statement instanceof InsertTabletStatement) {
120+
statements.add(statement);
121+
} else if (statement instanceof InsertRowsStatement) {
122+
tableModelDatabaseInsertRowStatementMap
123+
.computeIfAbsent(statement.getDatabaseName().get(), k -> new ArrayList<>())
124+
.addAll(((InsertRowsStatement) statement).getInsertRowStatementList());
125+
} else {
126+
throw new UnsupportedOperationException(
127+
String.format(
128+
"unknown InsertBaseStatement %s constructed from PipeTransferTabletBinaryReqV2.",
129+
insertNodeReq));
130+
}
97131
continue;
98132
}
99133
if (statement instanceof InsertRowStatement) {
@@ -131,6 +165,16 @@ public List<InsertBaseStatement> constructStatements() {
131165
if (!insertMultiTabletsStatement.isEmpty()) {
132166
statements.add(insertMultiTabletsStatement);
133167
}
168+
169+
for (final Map.Entry<String, List<InsertRowStatement>> insertRows :
170+
tableModelDatabaseInsertRowStatementMap.entrySet()) {
171+
final InsertRowsStatement statement = new InsertRowsStatement();
172+
statement.setWriteToTable(true);
173+
statement.setDatabaseName(insertRows.getKey());
174+
statement.setInsertRowStatementList(insertRows.getValue());
175+
statements.add(statement);
176+
}
177+
134178
return statements;
135179
}
136180

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeTabletEventSorter.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -153,7 +153,7 @@ protected Object reorderValueListAndBitMap(
153153
private int getLastNonnullIndex(
154154
final int i, final BitMap originalBitMap, final BitMap deDuplicatedBitMap) {
155155
if (deDuplicatedIndex == null) {
156-
if (originalBitMap.isMarked(index[i])) {
156+
if (originalBitMap != null && originalBitMap.isMarked(index[i])) {
157157
deDuplicatedBitMap.mark(i);
158158
}
159159
return index[i];

0 commit comments

Comments
 (0)