Skip to content

Load: Optimized the downgraded logic for tsFile to insert more data when tsFile corrupted #17674

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 6 commits into from
Jun 23, 2026
Merged

Load: Optimized the downgraded logic for tsFile to insert more data when tsFile corrupted #17674

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ea1f8b2
down
Caideyipi May 14, 2026
a8d003c
Update LoadTreeStatementDataTypeConvertExecutionVisitorTest.java
Caideyipi May 14, 2026
d893fea
Merge branch 'master' of https://github.com/apache/iotdb into load-query
Caideyipi Jun 3, 2026
32ab5aa
Fix query parser deferred tablet failure handling
Caideyipi Jun 3, 2026
732bf87
Fix aligned load fallback corruption test
Caideyipi Jun 5, 2026
a3e00d5
Address load fallback review comments
Caideyipi Jun 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 101 additions & 7 deletions ...pache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,9 +60,9 @@
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
Expand All @@ -81,6 +81,7 @@
private final Map<IDeviceID, Boolean> deviceIsAlignedMap;
private final Map<String, TSDataType> measurementDataTypeMap;
private final TabletStringInternPool tabletStringInternPool = new TabletStringInternPool();
private RuntimeException deferredException;

@TestOnly
public TsFileInsertionEventQueryParser(
Expand Down Expand Up @@ -116,10 +117,41 @@
null,
false,
null,
null,
isWithMod);
}

public TsFileInsertionEventQueryParser(
final String pipeName,
final long creationTime,
final File tsFile,
final TreePattern pattern,
final long startTime,
final long endTime,
final PipeTaskMeta pipeTaskMeta,
final PipeInsertionEvent sourceEvent,
final IAuditEntity entity,
final boolean skipIfNoPrivileges,
final Map<IDeviceID, Boolean> deviceIsAlignedMap,
final boolean isWithMod)
throws IOException, IllegalPathException {
this(
pipeName,
creationTime,
tsFile,
pattern,
startTime,
endTime,
pipeTaskMeta,
sourceEvent,
entity,
skipIfNoPrivileges,
deviceIsAlignedMap,
null,
isWithMod);
}

public TsFileInsertionEventQueryParser(

Check warning on line 154 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Constructor has 13 parameters, which is greater than 7 authorized. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ7uoz8Z8fB74yYqVBQy&open=AZ7uoz8Z8fB74yYqVBQy&pullRequest=17674
final String pipeName,
final long creationTime,
final File tsFile,
Expand All @@ -131,6 +163,7 @@
final IAuditEntity entity,
final boolean skipIfNoPrivileges,
final Map<IDeviceID, Boolean> deviceIsAlignedMap,
final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
final boolean isWithMod)
throws IOException, IllegalPathException {
super(
Expand Down Expand Up @@ -165,7 +198,25 @@
tsFileSequenceReader = new TsFileSequenceReader(tsFile.getPath(), true, true);
tsFileReader = new TsFileReader(tsFileSequenceReader);

if (tsFileResourceManager.cacheObjectsIfAbsent(tsFile)) {
if (Objects.nonNull(deviceMeasurementsMapOverride)) {
this.deviceIsAlignedMap =
Objects.nonNull(deviceIsAlignedMap)
? new LinkedHashMap<>(deviceIsAlignedMap)
: readDeviceIsAlignedMap();
memoryRequiredInBytes +=
Objects.nonNull(deviceIsAlignedMap)
? 0
: PipeMemoryWeightUtil.memoryOfIDeviceId2Bool(this.deviceIsAlignedMap);

measurementDataTypeMap =
readFilteredFullPathDataTypeMap(deviceMeasurementsMapOverride.keySet());
memoryRequiredInBytes +=
PipeMemoryWeightUtil.memoryOfStr2TSDataType(measurementDataTypeMap);

deviceMeasurementsMap = new LinkedHashMap<>(deviceMeasurementsMapOverride);
memoryRequiredInBytes +=
PipeMemoryWeightUtil.memoryOfIDeviceID2StrList(deviceMeasurementsMap);
} else if (tsFileResourceManager.cacheObjectsIfAbsent(tsFile)) {
// These read-only objects can be found in cache.
this.deviceIsAlignedMap =
Objects.nonNull(deviceIsAlignedMap)
Expand Down Expand Up @@ -266,10 +317,34 @@
}
}

public TsFileInsertionEventQueryParser(
final File tsFile,
final TreePattern pattern,
final long startTime,
final long endTime,
final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
final boolean isWithMod)
throws IOException, IllegalPathException {
this(
null,
0,
tsFile,
pattern,
startTime,
endTime,
null,
null,
null,
false,
null,
deviceMeasurementsMapOverride,
isWithMod);
}

private Map<IDeviceID, List<String>> filterDeviceMeasurementsMapByPattern(
final Map<IDeviceID, List<String>> originalDeviceMeasurementsMap)
throws IllegalPathException {
final Map<IDeviceID, List<String>> filteredDeviceMeasurementsMap = new HashMap<>();
final Map<IDeviceID, List<String>> filteredDeviceMeasurementsMap = new LinkedHashMap<>();
for (Map.Entry<IDeviceID, List<String>> entry : originalDeviceMeasurementsMap.entrySet()) {
final IDeviceID deviceId = entry.getKey();

Expand Down Expand Up @@ -318,7 +393,7 @@
}

private Map<IDeviceID, Boolean> readDeviceIsAlignedMap() throws IOException {
final Map<IDeviceID, Boolean> deviceIsAlignedResultMap = new HashMap<>();
final Map<IDeviceID, Boolean> deviceIsAlignedResultMap = new LinkedHashMap<>();
final TsFileDeviceIterator deviceIsAlignedIterator =
tsFileSequenceReader.getAllDevicesIteratorWithIsAligned();
while (deviceIsAlignedIterator.hasNext()) {
Expand Down Expand Up @@ -348,7 +423,7 @@
*/
private Map<String, TSDataType> readFilteredFullPathDataTypeMap(final Set<IDeviceID> devices)
throws IOException {
final Map<String, TSDataType> result = new HashMap<>();
final Map<String, TSDataType> result = new LinkedHashMap<>();

for (final IDeviceID device : devices) {
tsFileSequenceReader
Expand All @@ -370,7 +445,7 @@
*/
private Map<IDeviceID, List<String>> readFilteredDeviceMeasurementsMap(
final Set<IDeviceID> devices) throws IOException {
final Map<IDeviceID, List<String>> result = new HashMap<>();
final Map<IDeviceID, List<String>> result = new LinkedHashMap<>();

for (final IDeviceID device : devices) {
tsFileSequenceReader
Expand All @@ -397,6 +472,7 @@

@Override
public boolean hasNext() {
throwIfDeferredException();
boolean hasNext = false;
while (tabletIterator == null || !tabletIterator.hasNext()) {
if (!deviceMeasurementsMapIterator.hasNext()) {
Expand Down Expand Up @@ -451,9 +527,16 @@
final boolean isAligned =
deviceIsAlignedMap.getOrDefault(
IDeviceID.Factory.DEFAULT_FACTORY.create(tablet.getDeviceId()), false);
boolean isLast;
try {
isLast = !hasNext();
} catch (final RuntimeException e) {
deferredException = e;
isLast = false;
}

final TabletInsertionEvent next;
if (!hasNext()) {
if (isLast) {
next =
sourceEvent == null
? new PipeRawTabletInsertionEvent(
Expand Down Expand Up @@ -517,8 +600,19 @@
return tabletInsertionIterable;
}

private void throwIfDeferredException() {
if (Objects.isNull(deferredException)) {
return;
}

final RuntimeException exception = deferredException;
deferredException = null;
throw exception;
}

@Override
public void close() {
deferredException = null;
try {
if (tsFileReader != null) {
tsFileReader.close();
Expand Down
57 changes: 50 additions & 7 deletions .../apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,7 @@
private boolean currentIsAligned;
private final List<IMeasurementSchema> currentMeasurements = new ArrayList<>();
private final TabletStringInternPool tabletStringInternPool = new TabletStringInternPool();
private Exception deferredException;
private final List<ModsOperationUtil.ModsInfo> modsInfos = new ArrayList<>();
// Cached time chunk
private final List<Chunk> timeChunkList = new ArrayList<>();
Expand Down Expand Up @@ -204,6 +205,7 @@

@Override
public boolean hasNext() {
throwIfDeferredException();
final boolean hasNext = Objects.nonNull(chunkReader);
if (hasNext && !parseStartTimeRecorded) {
// Record start time on first hasNext() that returns true
Expand Down Expand Up @@ -232,7 +234,7 @@
final Tablet tablet = getNextTablet();
// Record tablet metrics
recordTabletMetrics(tablet);
final boolean hasNext = hasNext();
final boolean isLast = isLastTabletWithoutDeferredException();
try {
return sourceEvent == null
? new PipeRawTabletInsertionEvent(
Expand All @@ -246,7 +248,7 @@
0,
pipeTaskMeta,
sourceEvent,
!hasNext)
isLast)
: new PipeRawTabletInsertionEvent(
sourceEvent.getRawIsTableModelEvent(),
sourceEvent.getSourceDatabaseNameFromDataRegion(),
Expand All @@ -258,9 +260,10 @@
sourceEvent.getCreationTime(),
pipeTaskMeta,
sourceEvent,
!hasNext);
isLast);
} finally {
if (!hasNext) {
if (isLast) {
recordParseEndTimeIfNecessary();
close();
}
}
Expand All @@ -275,6 +278,7 @@
new Iterator<Pair<Tablet, Boolean>>() {
@Override
public boolean hasNext() {
throwIfDeferredException();
return Objects.nonNull(chunkReader);
}

Expand All @@ -291,18 +295,33 @@
// information.
final boolean isAligned = currentIsAligned;
final Tablet tablet = getNextTablet();
final boolean hasNext = hasNext();
try {
return new Pair<>(tablet, isAligned);
} finally {
if (!hasNext) {
if (isLastTabletWithoutDeferredException()) {
close();
}
}
}
};
}

public IDeviceID getCurrentDevice() {
return currentDevice;
}

public boolean isCurrentAligned() {
return currentIsAligned;
}

public List<String> getCurrentMeasurements() {
final List<String> measurementIds = new ArrayList<>(currentMeasurements.size());
for (final IMeasurementSchema schema : currentMeasurements) {
measurementIds.add(schema.getMeasurementName());
}
return measurementIds;
}

private Tablet getNextTablet() {
try {
Tablet tablet = null;
Expand Down Expand Up @@ -354,7 +373,11 @@

// Switch chunk reader iff current chunk is all consumed
if (!data.hasCurrent()) {
prepareData();
try {

Check warning on line 376 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Extract this nested try block into a separate method. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ7uoz8B8fB74yYqVBQw&open=AZ7uoz8B8fB74yYqVBQw&pullRequest=17674
prepareData();
} catch (final Exception e) {
deferredException = e;
}
}
PipeTabletUtils.compactBitMaps(tablet);
return tablet;
Expand All @@ -364,6 +387,26 @@
}
}

private void throwIfDeferredException() {
if (Objects.isNull(deferredException)) {
return;
}

final Exception exception = deferredException;
deferredException = null;
throw new PipeException("Failed to prepare next tablet insertion event.", exception);
}

private boolean isLastTabletWithoutDeferredException() {
return Objects.isNull(deferredException) && Objects.isNull(chunkReader);
}

private void recordParseEndTimeIfNecessary() {
if (parseStartTimeRecorded && !parseEndTimeRecorded) {
recordParseEndTime();
}
}

private void prepareData() throws IOException, IllegalPathException {
do {
do {
Expand Down Expand Up @@ -405,7 +448,7 @@
}
}

private boolean putValueToColumns(final BatchData data, final Tablet tablet, final int rowIndex) {

Check warning on line 451 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

A "Brain Method" was detected. Refactor it to reduce at least one of the following metrics: LOC from 134 to 64, Complexity from 34 to 14, Nesting Level from 4 to 2, Number of Variables from 8 to 6. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ7uoz8B8fB74yYqVBQx&open=AZ7uoz8B8fB74yYqVBQx&pullRequest=17674
boolean isNeedFillTime = false;
if (data.getDataType() == TSDataType.VECTOR) {
for (int i = 0; i < tablet.getSchemas().size(); ++i) {
Expand Down
15 changes: 3 additions & 12 deletions ...tdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,8 @@
package org.apache.iotdb.db.storageengine.load.converter;

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

try {
for (final File file : loadTsFileStatement.getTsFiles()) {
try (final TsFileInsertionEventScanParser parser =
new TsFileInsertionEventScanParser(
file,
new IoTDBTreePattern(null),
Long.MIN_VALUE,
Long.MAX_VALUE,
null,
null,
true)) {
for (final Pair<Tablet, Boolean> tabletWithIsAligned : parser.toTabletWithIsAligneds()) {
try (final LoadTreeTsFileTabletIterator tabletIterator =
Comment thread
Caideyipi marked this conversation as resolved.
new LoadTreeTsFileTabletIterator(file, true)) {
for (final Pair<Tablet, Boolean> tabletWithIsAligned : tabletIterator) {
final PipeTransferTabletRawReq tabletRawReq =
PipeTransferTabletRawReq.toTPipeTransferRawReq(
tabletWithIsAligned.getLeft(), tabletWithIsAligned.getRight());
Expand Down
Loading
Loading