Skip to content

Commit cb97fe4

Browse files
CaideyipiCaideyipi
authored
Load: Optimized the downgraded logic for tsFile to insert more data when tsFile corrupted (#17674)
* down * Update LoadTreeStatementDataTypeConvertExecutionVisitorTest.java * Fix query parser deferred tablet failure handling * Fix aligned load fallback corruption test * Address load fallback review comments
1 parent 511d08f commit cb97fe4

5 files changed

Lines changed: 1091 additions & 26 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java‎

Lines changed: 101 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -60,9 +60,9 @@
6060
import java.io.IOException;
6161
import java.util.ArrayList;
6262
import java.util.Collections;
63-
import java.util.HashMap;
6463
import java.util.HashSet;
6564
import java.util.Iterator;
65+
import java.util.LinkedHashMap;
6666
import java.util.List;
6767
import java.util.Map;
6868
import java.util.NoSuchElementException;
@@ -81,6 +81,7 @@ public class TsFileInsertionEventQueryParser extends TsFileInsertionEventParser
8181
private final Map<IDeviceID, Boolean> deviceIsAlignedMap;
8282
private final Map<String, TSDataType> measurementDataTypeMap;
8383
private final TabletStringInternPool tabletStringInternPool = new TabletStringInternPool();
84+
private RuntimeException deferredException;
8485

8586
@TestOnly
8687
public TsFileInsertionEventQueryParser(
@@ -116,6 +117,37 @@ public TsFileInsertionEventQueryParser(
116117
null,
117118
false,
118119
null,
120+
null,
121+
isWithMod);
122+
}
123+
124+
public TsFileInsertionEventQueryParser(
125+
final String pipeName,
126+
final long creationTime,
127+
final File tsFile,
128+
final TreePattern pattern,
129+
final long startTime,
130+
final long endTime,
131+
final PipeTaskMeta pipeTaskMeta,
132+
final PipeInsertionEvent sourceEvent,
133+
final IAuditEntity entity,
134+
final boolean skipIfNoPrivileges,
135+
final Map<IDeviceID, Boolean> deviceIsAlignedMap,
136+
final boolean isWithMod)
137+
throws IOException, IllegalPathException {
138+
this(
139+
pipeName,
140+
creationTime,
141+
tsFile,
142+
pattern,
143+
startTime,
144+
endTime,
145+
pipeTaskMeta,
146+
sourceEvent,
147+
entity,
148+
skipIfNoPrivileges,
149+
deviceIsAlignedMap,
150+
null,
119151
isWithMod);
120152
}
121153

@@ -131,6 +163,7 @@ public TsFileInsertionEventQueryParser(
131163
final IAuditEntity entity,
132164
final boolean skipIfNoPrivileges,
133165
final Map<IDeviceID, Boolean> deviceIsAlignedMap,
166+
final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
134167
final boolean isWithMod)
135168
throws IOException, IllegalPathException {
136169
super(
@@ -165,7 +198,25 @@ public TsFileInsertionEventQueryParser(
165198
tsFileSequenceReader = new TsFileSequenceReader(tsFile.getPath(), true, true);
166199
tsFileReader = new TsFileReader(tsFileSequenceReader);
167200

168-
if (tsFileResourceManager.cacheObjectsIfAbsent(tsFile)) {
201+
if (Objects.nonNull(deviceMeasurementsMapOverride)) {
202+
this.deviceIsAlignedMap =
203+
Objects.nonNull(deviceIsAlignedMap)
204+
? new LinkedHashMap<>(deviceIsAlignedMap)
205+
: readDeviceIsAlignedMap();
206+
memoryRequiredInBytes +=
207+
Objects.nonNull(deviceIsAlignedMap)
208+
? 0
209+
: PipeMemoryWeightUtil.memoryOfIDeviceId2Bool(this.deviceIsAlignedMap);
210+
211+
measurementDataTypeMap =
212+
readFilteredFullPathDataTypeMap(deviceMeasurementsMapOverride.keySet());
213+
memoryRequiredInBytes +=
214+
PipeMemoryWeightUtil.memoryOfStr2TSDataType(measurementDataTypeMap);
215+
216+
deviceMeasurementsMap = new LinkedHashMap<>(deviceMeasurementsMapOverride);
217+
memoryRequiredInBytes +=
218+
PipeMemoryWeightUtil.memoryOfIDeviceID2StrList(deviceMeasurementsMap);
219+
} else if (tsFileResourceManager.cacheObjectsIfAbsent(tsFile)) {
169220
// These read-only objects can be found in cache.
170221
this.deviceIsAlignedMap =
171222
Objects.nonNull(deviceIsAlignedMap)
@@ -266,10 +317,34 @@ public TsFileInsertionEventQueryParser(
266317
}
267318
}
268319

320+
public TsFileInsertionEventQueryParser(
321+
final File tsFile,
322+
final TreePattern pattern,
323+
final long startTime,
324+
final long endTime,
325+
final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
326+
final boolean isWithMod)
327+
throws IOException, IllegalPathException {
328+
this(
329+
null,
330+
0,
331+
tsFile,
332+
pattern,
333+
startTime,
334+
endTime,
335+
null,
336+
null,
337+
null,
338+
false,
339+
null,
340+
deviceMeasurementsMapOverride,
341+
isWithMod);
342+
}
343+
269344
private Map<IDeviceID, List<String>> filterDeviceMeasurementsMapByPattern(
270345
final Map<IDeviceID, List<String>> originalDeviceMeasurementsMap)
271346
throws IllegalPathException {
272-
final Map<IDeviceID, List<String>> filteredDeviceMeasurementsMap = new HashMap<>();
347+
final Map<IDeviceID, List<String>> filteredDeviceMeasurementsMap = new LinkedHashMap<>();
273348
for (Map.Entry<IDeviceID, List<String>> entry : originalDeviceMeasurementsMap.entrySet()) {
274349
final IDeviceID deviceId = entry.getKey();
275350

@@ -318,7 +393,7 @@ else if (treePattern.mayOverlapWithDevice(deviceId)) {
318393
}
319394

320395
private Map<IDeviceID, Boolean> readDeviceIsAlignedMap() throws IOException {
321-
final Map<IDeviceID, Boolean> deviceIsAlignedResultMap = new HashMap<>();
396+
final Map<IDeviceID, Boolean> deviceIsAlignedResultMap = new LinkedHashMap<>();
322397
final TsFileDeviceIterator deviceIsAlignedIterator =
323398
tsFileSequenceReader.getAllDevicesIteratorWithIsAligned();
324399
while (deviceIsAlignedIterator.hasNext()) {
@@ -348,7 +423,7 @@ private Set<IDeviceID> filterDevicesByPattern(final Set<IDeviceID> devices) {
348423
*/
349424
private Map<String, TSDataType> readFilteredFullPathDataTypeMap(final Set<IDeviceID> devices)
350425
throws IOException {
351-
final Map<String, TSDataType> result = new HashMap<>();
426+
final Map<String, TSDataType> result = new LinkedHashMap<>();
352427

353428
for (final IDeviceID device : devices) {
354429
tsFileSequenceReader
@@ -370,7 +445,7 @@ private Map<String, TSDataType> readFilteredFullPathDataTypeMap(final Set<IDevic
370445
*/
371446
private Map<IDeviceID, List<String>> readFilteredDeviceMeasurementsMap(
372447
final Set<IDeviceID> devices) throws IOException {
373-
final Map<IDeviceID, List<String>> result = new HashMap<>();
448+
final Map<IDeviceID, List<String>> result = new LinkedHashMap<>();
374449

375450
for (final IDeviceID device : devices) {
376451
tsFileSequenceReader
@@ -397,6 +472,7 @@ public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
397472

398473
@Override
399474
public boolean hasNext() {
475+
throwIfDeferredException();
400476
boolean hasNext = false;
401477
while (tabletIterator == null || !tabletIterator.hasNext()) {
402478
if (!deviceMeasurementsMapIterator.hasNext()) {
@@ -451,9 +527,16 @@ public TabletInsertionEvent next() {
451527
final boolean isAligned =
452528
deviceIsAlignedMap.getOrDefault(
453529
IDeviceID.Factory.DEFAULT_FACTORY.create(tablet.getDeviceId()), false);
530+
boolean isLast;
531+
try {
532+
isLast = !hasNext();
533+
} catch (final RuntimeException e) {
534+
deferredException = e;
535+
isLast = false;
536+
}
454537

455538
final TabletInsertionEvent next;
456-
if (!hasNext()) {
539+
if (isLast) {
457540
next =
458541
sourceEvent == null
459542
? new PipeRawTabletInsertionEvent(
@@ -517,8 +600,19 @@ public TabletInsertionEvent next() {
517600
return tabletInsertionIterable;
518601
}
519602

603+
private void throwIfDeferredException() {
604+
if (Objects.isNull(deferredException)) {
605+
return;
606+
}
607+
608+
final RuntimeException exception = deferredException;
609+
deferredException = null;
610+
throw exception;
611+
}
612+
520613
@Override
521614
public void close() {
615+
deferredException = null;
522616
try {
523617
if (tsFileReader != null) {
524618
tsFileReader.close();

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java‎

Lines changed: 50 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,7 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
9898
private boolean currentIsAligned;
9999
private final List<IMeasurementSchema> currentMeasurements = new ArrayList<>();
100100
private final TabletStringInternPool tabletStringInternPool = new TabletStringInternPool();
101+
private Exception deferredException;
101102
private final List<ModsOperationUtil.ModsInfo> modsInfos = new ArrayList<>();
102103
// Cached time chunk
103104
private final List<Chunk> timeChunkList = new ArrayList<>();
@@ -204,6 +205,7 @@ public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
204205

205206
@Override
206207
public boolean hasNext() {
208+
throwIfDeferredException();
207209
final boolean hasNext = Objects.nonNull(chunkReader);
208210
if (hasNext && !parseStartTimeRecorded) {
209211
// Record start time on first hasNext() that returns true
@@ -232,7 +234,7 @@ public TabletInsertionEvent next() {
232234
final Tablet tablet = getNextTablet();
233235
// Record tablet metrics
234236
recordTabletMetrics(tablet);
235-
final boolean hasNext = hasNext();
237+
final boolean isLast = isLastTabletWithoutDeferredException();
236238
try {
237239
return sourceEvent == null
238240
? new PipeRawTabletInsertionEvent(
@@ -246,7 +248,7 @@ public TabletInsertionEvent next() {
246248
0,
247249
pipeTaskMeta,
248250
sourceEvent,
249-
!hasNext)
251+
isLast)
250252
: new PipeRawTabletInsertionEvent(
251253
sourceEvent.getRawIsTableModelEvent(),
252254
sourceEvent.getSourceDatabaseNameFromDataRegion(),
@@ -258,9 +260,10 @@ public TabletInsertionEvent next() {
258260
sourceEvent.getCreationTime(),
259261
pipeTaskMeta,
260262
sourceEvent,
261-
!hasNext);
263+
isLast);
262264
} finally {
263-
if (!hasNext) {
265+
if (isLast) {
266+
recordParseEndTimeIfNecessary();
264267
close();
265268
}
266269
}
@@ -275,6 +278,7 @@ public Iterable<Pair<Tablet, Boolean>> toTabletWithIsAligneds() {
275278
new Iterator<Pair<Tablet, Boolean>>() {
276279
@Override
277280
public boolean hasNext() {
281+
throwIfDeferredException();
278282
return Objects.nonNull(chunkReader);
279283
}
280284

@@ -291,18 +295,33 @@ public Pair<Tablet, Boolean> next() {
291295
// information.
292296
final boolean isAligned = currentIsAligned;
293297
final Tablet tablet = getNextTablet();
294-
final boolean hasNext = hasNext();
295298
try {
296299
return new Pair<>(tablet, isAligned);
297300
} finally {
298-
if (!hasNext) {
301+
if (isLastTabletWithoutDeferredException()) {
299302
close();
300303
}
301304
}
302305
}
303306
};
304307
}
305308

309+
public IDeviceID getCurrentDevice() {
310+
return currentDevice;
311+
}
312+
313+
public boolean isCurrentAligned() {
314+
return currentIsAligned;
315+
}
316+
317+
public List<String> getCurrentMeasurements() {
318+
final List<String> measurementIds = new ArrayList<>(currentMeasurements.size());
319+
for (final IMeasurementSchema schema : currentMeasurements) {
320+
measurementIds.add(schema.getMeasurementName());
321+
}
322+
return measurementIds;
323+
}
324+
306325
private Tablet getNextTablet() {
307326
try {
308327
Tablet tablet = null;
@@ -354,7 +373,11 @@ private Tablet getNextTablet() {
354373

355374
// Switch chunk reader iff current chunk is all consumed
356375
if (!data.hasCurrent()) {
357-
prepareData();
376+
try {
377+
prepareData();
378+
} catch (final Exception e) {
379+
deferredException = e;
380+
}
358381
}
359382
PipeTabletUtils.compactBitMaps(tablet);
360383
return tablet;
@@ -364,6 +387,26 @@ private Tablet getNextTablet() {
364387
}
365388
}
366389

390+
private void throwIfDeferredException() {
391+
if (Objects.isNull(deferredException)) {
392+
return;
393+
}
394+
395+
final Exception exception = deferredException;
396+
deferredException = null;
397+
throw new PipeException("Failed to prepare next tablet insertion event.", exception);
398+
}
399+
400+
private boolean isLastTabletWithoutDeferredException() {
401+
return Objects.isNull(deferredException) && Objects.isNull(chunkReader);
402+
}
403+
404+
private void recordParseEndTimeIfNecessary() {
405+
if (parseStartTimeRecorded && !parseEndTimeRecorded) {
406+
recordParseEndTime();
407+
}
408+
}
409+
367410
private void prepareData() throws IOException, IllegalPathException {
368411
do {
369412
do {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java‎

Lines changed: 3 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -20,10 +20,8 @@
2020
package org.apache.iotdb.db.storageengine.load.converter;
2121

2222
import org.apache.iotdb.common.rpc.thrift.TSStatus;
23-
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
2423
import org.apache.iotdb.db.conf.IoTDBDescriptor;
2524
import org.apache.iotdb.db.i18n.StorageEngineMessages;
26-
import org.apache.iotdb.db.pipe.event.common.tsfile.parser.scan.TsFileInsertionEventScanParser;
2725
import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletRawReq;
2826
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
2927
import org.apache.iotdb.db.queryengine.plan.statement.StatementNode;
@@ -90,16 +88,9 @@ public Optional<TSStatus> visitLoadFile(
9088

9189
try {
9290
for (final File file : loadTsFileStatement.getTsFiles()) {
93-
try (final TsFileInsertionEventScanParser parser =
94-
new TsFileInsertionEventScanParser(
95-
file,
96-
new IoTDBTreePattern(null),
97-
Long.MIN_VALUE,
98-
Long.MAX_VALUE,
99-
null,
100-
null,
101-
true)) {
102-
for (final Pair<Tablet, Boolean> tabletWithIsAligned : parser.toTabletWithIsAligneds()) {
91+
try (final LoadTreeTsFileTabletIterator tabletIterator =
92+
new LoadTreeTsFileTabletIterator(file, true)) {
93+
for (final Pair<Tablet, Boolean> tabletWithIsAligned : tabletIterator) {
10394
final PipeTransferTabletRawReq tabletRawReq =
10495
PipeTransferTabletRawReq.toTPipeTransferRawReq(
10596
tabletWithIsAligned.getLeft(), tabletWithIsAligned.getRight());

0 commit comments

Comments
 (0)