This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new bdf6b7ecfe9 [To dev/1.3] Load: preserve pending tablets when parser
fails (#18487) (#18509)
bdf6b7ecfe9 is described below
commit bdf6b7ecfe905a26b1fb2030daa8c97278438008
Author: Caideyipi <[email protected]>
AuthorDate: Tue Aug 25 14:21:06 2026 +0800
[To dev/1.3] Load: preserve pending tablets when parser fails (#18487)
(#18509)
* Load: preserve pending tablets when parser fails (#18487)
(cherry picked from commit 429587f629a597e63d47a026258a308b2dbee6ac)
* Fix pending tablet load test row count
---
...eeStatementDataTypeConvertExecutionVisitor.java | 64 +++++++++++++++++++++-
...atementDataTypeConvertExecutionVisitorTest.java | 60 ++++++++++++++++++++
2 files changed, 121 insertions(+), 3 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java
index 03e13f01dca..e5709a47265 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java
@@ -43,6 +43,7 @@ import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
+import java.util.function.Function;
import java.util.stream.Collectors;
import static
org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil.calculateTabletSizeInBytes;
@@ -58,6 +59,7 @@ public class LoadTreeStatementDataTypeConvertExecutionVisitor
.getLoadTsFileTabletConversionBatchMemorySizeInBytes();
private final StatementExecutor statementExecutor;
+ private final Function<File, LoadTreeTsFileTabletIterator>
tabletIteratorFactory;
@FunctionalInterface
public interface StatementExecutor {
@@ -66,7 +68,14 @@ public class LoadTreeStatementDataTypeConvertExecutionVisitor
public LoadTreeStatementDataTypeConvertExecutionVisitor(
final StatementExecutor statementExecutor) {
+ this(statementExecutor, file -> new LoadTreeTsFileTabletIterator(file,
true));
+ }
+
+ LoadTreeStatementDataTypeConvertExecutionVisitor(
+ final StatementExecutor statementExecutor,
+ final Function<File, LoadTreeTsFileTabletIterator>
tabletIteratorFactory) {
this.statementExecutor = statementExecutor;
+ this.tabletIteratorFactory = tabletIteratorFactory;
}
@Override
@@ -88,11 +97,24 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitor
boolean shouldReleaseContext = !isManagedTask;
try {
+ if (conversionContext.deferredStatus != null) {
+ final TSStatus result =
+ flushPendingTablets(conversionContext,
loadTsFileStatement.isConvertOnTypeMismatch());
+ if (!handleTSStatus(result, loadTsFileStatement)) {
+ shouldReleaseContext = !isManagedTask ||
!isTemporaryUnavailable(result);
+ return Optional.of(result);
+ }
+ final TSStatus deferredStatus = conversionContext.deferredStatus;
+ conversionContext.deferredStatus = null;
+ shouldReleaseContext = true;
+ return Optional.of(deferredStatus);
+ }
+
final List<File> files = loadTsFileStatement.getTsFiles();
while (conversionContext.fileIndex < files.size()) {
if (conversionContext.tabletIterator == null) {
conversionContext.tabletIterator =
- new
LoadTreeTsFileTabletIterator(files.get(conversionContext.fileIndex), true);
+
tabletIteratorFactory.apply(files.get(conversionContext.fileIndex));
}
if (conversionContext.deferredTabletRawReq != null) {
@@ -153,8 +175,27 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitor
"Failed to convert data type for LoadTsFileStatement: {}.",
loadTsFileStatement, e);
final TSStatus status =
loadTsFileStatement.accept(LoadTsFileDataTypeConverter.STATEMENT_EXCEPTION_VISITOR,
e);
- shouldReleaseContext =
- !isManagedTask ||
!LoadTsFileDataTypeConverter.isMemoryPressureException(e);
+
+ final boolean retryable = isRetryableConversionException(e);
+
+ // A parser can fail after producing tablets that are still waiting for
the next batch
+ // boundary. Submit those tablets before reporting the parser error so
the error does not
+ // discard successfully converted data.
+ if (!retryable && !conversionContext.tabletRawReqs.isEmpty()) {
+ final TSStatus flushStatus =
+ flushPendingTablets(conversionContext,
loadTsFileStatement.isConvertOnTypeMismatch());
+ if (!handleTSStatus(flushStatus, loadTsFileStatement)) {
+ if (isManagedTask && isTemporaryUnavailable(flushStatus)) {
+ conversionContext.deferredStatus = status;
+ shouldReleaseContext = false;
+ } else {
+ shouldReleaseContext = true;
+ }
+ return Optional.of(flushStatus);
+ }
+ }
+
+ shouldReleaseContext = !isManagedTask || !retryable;
return Optional.of(status);
} finally {
if (shouldReleaseContext) {
@@ -205,6 +246,21 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitor
==
TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode());
}
+ private static boolean isRetryableConversionException(final Throwable
throwable) {
+ if (LoadTsFileDataTypeConverter.isMemoryPressureException(throwable)) {
+ return true;
+ }
+
+ Throwable current = throwable;
+ while (current != null) {
+ if (current instanceof InterruptedException) {
+ return true;
+ }
+ current = current.getCause();
+ }
+ return false;
+ }
+
private static final class TreeConversionContext implements AutoCloseable {
private final LoadTsFileMemoryBlock block =
LoadTsFileMemoryManager.getInstance()
@@ -216,6 +272,7 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitor
private Pair<Tablet, Boolean> deferredTabletWithIsAligned;
private PipeTransferTabletRawReq deferredTabletRawReq;
private long deferredTabletRawReqSize;
+ private TSStatus deferredStatus;
private void addTablet(final PipeTransferTabletRawReq request, final long
size) {
tabletRawReqs.add(request);
@@ -247,6 +304,7 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitor
deferredTabletWithIsAligned = null;
deferredTabletRawReq = null;
deferredTabletRawReqSize = 0;
+ deferredStatus = null;
block.close();
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitorTest.java
index 150c950fce3..0cfbbc364f4 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitorTest.java
@@ -38,6 +38,7 @@ import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.read.TsFileSequenceReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.write.TsFileWriter;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -110,6 +111,33 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
Assert.assertEquals(loadedPointCountBeforeCorruption,
loadedPointCountAfterFallback);
}
+ @Test
+ public void testFlushesPendingTabletsWhenIteratorFails() throws Exception {
+ tsFile = File.createTempFile("load-tree-pending-tablet", ".tsfile");
+ final List<MeasurementSchema> schemaList =
+ Arrays.asList(new MeasurementSchema("s0", TSDataType.INT64,
TSEncoding.PLAIN));
+ final Tablet tablet = new Tablet(DEVICE_0, schemaList, 1);
+ tablet.addTimestamp(0, 1);
+ tablet.addValue("s0", 0, 1L);
+ tablet.rowSize = 1;
+
+ final Map<String, Integer> pointCountByDevice = new HashMap<>();
+ final LoadTreeStatementDataTypeConvertExecutionVisitor visitor =
+ new LoadTreeStatementDataTypeConvertExecutionVisitor(
+ statement -> {
+ collectLoadedPoints(statement, pointCountByDevice);
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ },
+ file -> new ThrowingTabletIterator(file, tablet));
+
+ final Optional<TSStatus> status =
+
visitor.visitLoadFile(LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()),
null);
+
+ Assert.assertTrue(status.isPresent());
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
status.get().getCode());
+ Assert.assertEquals(1, pointCountByDevice.getOrDefault(DEVICE_0,
0).intValue());
+ }
+
@Test
public void testFallbackToQueryWhenFirstNonAlignedDeviceIsCorrupted() throws
Exception {
tsFile = new
File("load-tree-query-fallback-corrupted-first-non-aligned-device.tsfile");
@@ -393,4 +421,36 @@ public class
LoadTreeStatementDataTypeConvertExecutionVisitorTest {
CommonDescriptor.getInstance().getConfig().setPipeMemoryManagementEnabled(enabled);
}
}
+
+ private static class ThrowingTabletIterator extends
LoadTreeTsFileTabletIterator {
+ private final Pair<Tablet, Boolean> tabletWithIsAligned;
+ private boolean tabletAvailable = true;
+
+ private ThrowingTabletIterator(final File file, final Tablet tablet) {
+ super(file, true);
+ tabletWithIsAligned = new Pair<>(tablet, false);
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (tabletAvailable) {
+ return true;
+ }
+ throw new IllegalStateException("synthetic parser failure");
+ }
+
+ @Override
+ public Pair<Tablet, Boolean> next() {
+ if (!tabletAvailable) {
+ throw new IllegalStateException("synthetic parser failure");
+ }
+ tabletAvailable = false;
+ return tabletWithIsAligned;
+ }
+
+ @Override
+ public void close() {
+ // No parser resources are allocated by this test iterator.
+ }
+ }
}
