Skip to content

[Pipe] Support receiver-side async takeover for TsFile conversion #18421

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

Open
Caideyipi wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[Pipe] Support receiver-side async takeover for TsFile conversion #18421

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3e2ae1f
[Pipe] Support async TsFile conversion takeover
Caideyipi Aug 7, 2026
88bb0d6
[Pipe] Retry receiver-side TsFile load failures
Caideyipi Aug 7, 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
208 changes: 149 additions & 59 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 @@ -93,6 +93,8 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {

private IChunkReader chunkReader;
private BatchData data;
private BatchData pendingPageDataAfterMemoryPressure;
private Tablet pendingTabletAfterMemoryPressure;
private final PipeMemoryBlock allocatedMemoryBlockForBatchData;
private final PipeMemoryBlock allocatedMemoryBlockForChunk;
private PipeMemoryBlock allocatedMemoryBlockForTsFileInput;
Expand All @@ -115,6 +117,7 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
private CachedAlignedValueChunk cachedAlignedValueChunk;

private byte lastMarker = Byte.MIN_VALUE;
private boolean shouldRetryChunkHeaderAfterMemoryPressure;

public TsFileInsertionEventScanParser(
final String pipeName,
Expand Down Expand Up @@ -223,6 +226,7 @@ public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
@Override
public boolean hasNext() {
throwIfDeferredException();
retryChunkHeaderAfterMemoryPressureIfNecessary();
final boolean hasNext = Objects.nonNull(chunkReader);
if (hasNext && !parseStartTimeRecorded) {
// Record start time on first hasNext() that returns true
Expand All @@ -242,8 +246,11 @@ public TabletInsertionEvent next() {
}

// Release the previous parser-owned tablet buffer before allocating the next
// tablet.
releaseTabletMemoryBlock();
// tablet. A tablet retained after memory pressure still owns this block and must
// keep it until the retry has successfully returned the pending event.
if (pendingTabletAfterMemoryPressure == null) {
releaseTabletMemoryBlock();
}
// currentIsAligned is initialized when TsFileInsertionEventScanParser is
// constructed.
// When the getNextTablet function is called, currentIsAligned may be updated,
Expand Down Expand Up @@ -302,6 +309,7 @@ public Iterable<Pair<Tablet, Boolean>> toTabletWithIsAligneds() {
@Override
public boolean hasNext() {
throwIfDeferredException();
retryChunkHeaderAfterMemoryPressureIfNecessary();
return Objects.nonNull(chunkReader);
}

Expand Down Expand Up @@ -346,15 +354,25 @@ public List<String> getCurrentMeasurements() {
}

private Tablet getNextTablet() {
Tablet tablet = null;
boolean tabletMemoryReserved = false;
try {
Tablet tablet = null;
if (data != null && !data.hasCurrent() && pendingPageDataAfterMemoryPressure != null) {
data = nextPageData();
}
tablet = pendingTabletAfterMemoryPressure;
tabletMemoryReserved = tablet != null;
pendingTabletAfterMemoryPressure = null;

if (!data.hasCurrent()) {
tablet = new Tablet(currentDeviceString, currentMeasurements, 1);
return tablet;
if (tablet != null) {
PipeTabletUtils.compactBitMaps(tablet);
return tablet;
}
return new Tablet(currentDeviceString, currentMeasurements, 1);
}

boolean isFirstRow = true;
boolean isFirstRow = tablet == null;
while (data.hasCurrent()) {
if (currentIsMultiPage
|| data.currentTime() >= startTime && data.currentTime() <= endTime) {
Expand All @@ -370,6 +388,8 @@ private Tablet getNextTablet() {
PipeDataNodeResourceManager.memory()
.forceResize(allocatedMemoryBlockForTablet, rowCountAndMemorySize.getRight());
}
tabletMemoryReserved = true;
pendingTabletAfterMemoryPressure = tablet;
isFirstRow = false;
}

Expand Down Expand Up @@ -403,8 +423,10 @@ private Tablet getNextTablet() {
}
}
PipeTabletUtils.compactBitMaps(tablet);
pendingTabletAfterMemoryPressure = null;
return tablet;
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
pendingTabletAfterMemoryPressure = tabletMemoryReserved ? tablet : null;
// Keep the parser state so the caller can yield its parser slot and retry from the same
// unconsumed data after memory is available again.
throw e;
Expand All @@ -426,6 +448,21 @@ private void throwIfDeferredException() {
exception);
}

private void retryChunkHeaderAfterMemoryPressureIfNecessary() {
if (!shouldRetryChunkHeaderAfterMemoryPressure) {
return;
}
try {
prepareData();
shouldRetryChunkHeaderAfterMemoryPressure = false;
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
throw e;
} catch (final Exception e) {
close();
throw new PipeException(DataNodePipeMessages.FAILED_TO_GET_NEXT_TABLET_INSERTION_EVENT, e);
}
}

private boolean isLastTabletWithoutDeferredException() {
return Objects.isNull(deferredException) && Objects.isNull(chunkReader);
}
Expand Down Expand Up @@ -454,9 +491,24 @@ private void prepareData() throws IOException, IllegalPathException {
}

private BatchData nextPageData() throws IOException {
if (pendingPageDataAfterMemoryPressure != null) {
resizePageDataMemoryIfNeeded(
PipeMemoryWeightUtil.calculateBatchDataRamBytesUsed(pendingPageDataAfterMemoryPressure));
final BatchData pendingPageData = pendingPageDataAfterMemoryPressure;
pendingPageDataAfterMemoryPressure = null;
return pendingPageData;
}

resizePageDataMemoryForCurrentPageIfNeeded();
final BatchData nextData = chunkReader.nextPageData();
resizePageDataMemoryIfNeeded(PipeMemoryWeightUtil.calculateBatchDataRamBytesUsed(nextData));
try {
resizePageDataMemoryIfNeeded(PipeMemoryWeightUtil.calculateBatchDataRamBytesUsed(nextData));
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
// The chunk reader has already advanced. Retain the decoded page until its memory can be
// accounted for instead of advancing to the following page on retry.
pendingPageDataAfterMemoryPressure = nextData;
throw e;
}
return nextData;
}

Expand Down Expand Up @@ -653,9 +705,20 @@ private void moveToNextChunkReader()
break;
}

if (chunkHeader.getDataSize() > allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
PipeDataNodeResourceManager.memory()
.forceResize(allocatedMemoryBlockForChunk, chunkHeader.getDataSize());
try {
if (chunkHeader.getDataSize()
> allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
PipeDataNodeResourceManager.memory()
.forceResize(allocatedMemoryBlockForChunk, chunkHeader.getDataSize());
}
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
// The marker has already been consumed. Rewind to the start of the header and retain
// the marker so a retry does not interpret header bytes as a marker.
tsFileSequenceReader.position(currentChunkHeaderOffset + 1);
lastMarker = marker;
chunkReader = null;
shouldRetryChunkHeaderAfterMemoryPressure = true;
throw e;
}

Chunk chunk =
Expand Down Expand Up @@ -717,10 +780,18 @@ private void moveToNextChunkReader()
cachedAlignedValueChunk = null;
}

if (returnPendingAlignedChunkBeforeCaching(valueChunk)) {
return;
try {
if (returnPendingAlignedChunkBeforeCaching(valueChunk)) {
return;
}
cacheAlignedValueChunk(valueChunk);
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
// The value chunk has already been read from the file. Keep it as the next logical
// input so a retry neither skips it nor reads it twice.
cachedAlignedValueChunk = valueChunk;
shouldRetryChunkHeaderAfterMemoryPressure = true;
throw e;
}
cacheAlignedValueChunk(valueChunk);
break;
}
case MetaMarker.CHUNK_GROUP_HEADER:
Expand Down Expand Up @@ -837,56 +908,70 @@ private boolean filterChunk(
}

private boolean useNextPendingAlignedChunk(final byte marker) throws IOException {
while (!pendingAlignedChunkGroups.isEmpty()) {
final PendingAlignedChunkGroup pendingAlignedChunkGroup = pendingAlignedChunkGroups.remove(0);
pendingAlignedChunkSize =
Math.max(0, pendingAlignedChunkSize - pendingAlignedChunkGroup.chunkSize);

if (pendingAlignedChunkGroup.valueChunkList.isEmpty()) {
continue;
}
try {
while (!pendingAlignedChunkGroups.isEmpty()) {
final PendingAlignedChunkGroup pendingAlignedChunkGroup = pendingAlignedChunkGroups.get(0);

final Chunk timeChunk = timeChunkList.get(pendingAlignedChunkGroup.timeChunkIndex);
timeChunk.getData().rewind();
for (final Chunk valueChunk : pendingAlignedChunkGroup.valueChunkList) {
valueChunk.getData().rewind();
}
if (pendingAlignedChunkGroup.valueChunkList.isEmpty()) {
pendingAlignedChunkGroups.remove(0);
pendingAlignedChunkSize =
Math.max(0, pendingAlignedChunkSize - pendingAlignedChunkGroup.chunkSize);
continue;
}

currentMeasurements.clear();
currentMeasurements.addAll(pendingAlignedChunkGroup.measurements);
modsInfos.clear();
modsInfos.addAll(pendingAlignedChunkGroup.modsInfos);
final Chunk timeChunk = timeChunkList.get(pendingAlignedChunkGroup.timeChunkIndex);
timeChunk.getData().rewind();
for (final Chunk valueChunk : pendingAlignedChunkGroup.valueChunkList) {
valueChunk.getData().rewind();
}

currentIsMultiPage = isMultiPageList.get(pendingAlignedChunkGroup.timeChunkIndex);
if (!currentIsMultiPage) {
resizePageDataMemoryIfNeeded(
AlignedSinglePageWholeChunkReader.calculatePageEstimatedMemoryUsageInBytes(
timeChunk, pendingAlignedChunkGroup.valueChunkList));
}
final List<Long> pageEstimatedMemoryUsageInBytesList =
currentIsMultiPage
? AlignedSinglePageWholeChunkReader
.calculatePageEstimatedMemoryUsageInBytesWithBatchDataList(
timeChunk, pendingAlignedChunkGroup.valueChunkList)
: Collections.emptyList();
final long maxPageEstimatedMemoryUsageInBytes =
pageEstimatedMemoryUsageInBytesList.isEmpty()
? 0
: pageEstimatedMemoryUsageInBytesList.get(0);
resizePageDataMemoryIfNeeded(maxPageEstimatedMemoryUsageInBytes);
chunkReader =
currentIsMultiPage
? new MemoryControlledChunkReader(
new AlignedChunkReader(
timeChunk, pendingAlignedChunkGroup.valueChunkList, filter),
pageEstimatedMemoryUsageInBytesList)
: new AlignedSinglePageWholeChunkReader(
timeChunk, pendingAlignedChunkGroup.valueChunkList, null);
currentIsAligned = true;
if (marker != Byte.MIN_VALUE) {
lastMarker = marker;
final boolean nextIsMultiPage =
isMultiPageList.get(pendingAlignedChunkGroup.timeChunkIndex);
if (!nextIsMultiPage) {
resizePageDataMemoryIfNeeded(
AlignedSinglePageWholeChunkReader.calculatePageEstimatedMemoryUsageInBytes(
timeChunk, pendingAlignedChunkGroup.valueChunkList));
}
final List<Long> pageEstimatedMemoryUsageInBytesList =
nextIsMultiPage
? AlignedSinglePageWholeChunkReader
.calculatePageEstimatedMemoryUsageInBytesWithBatchDataList(
timeChunk, pendingAlignedChunkGroup.valueChunkList)
: Collections.emptyList();
final long maxPageEstimatedMemoryUsageInBytes =
pageEstimatedMemoryUsageInBytesList.isEmpty()
? 0
: pageEstimatedMemoryUsageInBytesList.get(0);
resizePageDataMemoryIfNeeded(maxPageEstimatedMemoryUsageInBytes);
final IChunkReader nextChunkReader =
nextIsMultiPage
? new MemoryControlledChunkReader(
new AlignedChunkReader(
timeChunk, pendingAlignedChunkGroup.valueChunkList, filter),
pageEstimatedMemoryUsageInBytesList)
: new AlignedSinglePageWholeChunkReader(
timeChunk, pendingAlignedChunkGroup.valueChunkList, null);

// Publish the state transition only after all memory reservations and reader construction
// succeed. Otherwise a retry would lose the already-read aligned chunk group.
pendingAlignedChunkGroups.remove(0);
pendingAlignedChunkSize =
Math.max(0, pendingAlignedChunkSize - pendingAlignedChunkGroup.chunkSize);
currentMeasurements.clear();
currentMeasurements.addAll(pendingAlignedChunkGroup.measurements);
modsInfos.clear();
modsInfos.addAll(pendingAlignedChunkGroup.modsInfos);
currentIsMultiPage = nextIsMultiPage;
chunkReader = nextChunkReader;
currentIsAligned = true;
if (marker != Byte.MIN_VALUE) {
lastMarker = marker;
}
return true;
}
return true;
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
shouldRetryChunkHeaderAfterMemoryPressure = true;
throw e;
}
return false;
}
Expand Down Expand Up @@ -1114,6 +1199,11 @@ private byte toValueChunkMarker(final ChunkHeader chunkHeader) {
@Override
public void close() {
super.close();
pendingPageDataAfterMemoryPressure = null;
pendingTabletAfterMemoryPressure = null;
cachedAlignedValueChunk = null;
pendingAlignedChunkGroups.clear();
pendingAlignedChunkSize = 0;

if (allocatedMemoryBlockForBatchData != null) {
allocatedMemoryBlockForBatchData.close();
Expand Down
20 changes: 16 additions & 4 deletions ...pache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import org.apache.iotdb.db.pipe.event.common.tsfile.parser.util.ModsOperationUtil;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter;
import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeException;
Expand Down Expand Up @@ -136,6 +137,7 @@ public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {

private TsFileInsertionEventTableParserTabletIterator tabletIterator;
private PipeRawTabletInsertionEvent nextEvent;
private Tablet currentTablet;
private Tablet bufferedTablet;
private boolean iterationClosed = false;

Expand All @@ -146,15 +148,21 @@ public boolean hasNext() {
return true;
}

final Tablet tablet = pollNextNonEmptyTablet();
if (tablet == null) {
if (currentTablet == null) {
currentTablet = pollNextNonEmptyTablet();
}
if (currentTablet == null) {
return false;
}

nextEvent = buildTabletInsertionEvent(tablet, !prepareNextNonEmptyTablet());
nextEvent =
buildTabletInsertionEvent(currentTablet, !prepareNextNonEmptyTablet());
currentTablet = null;
return true;
} catch (Exception e) {
close();
if (!isMemoryPressureException(e)) {
close();
}
throw new PipeException(
DataNodePipeMessages.ERROR_WHILE_PARSING_TSFILE_INSERTION_EVENT, e);
}
Expand Down Expand Up @@ -295,6 +303,10 @@ public TabletInsertionEvent next() {
return tabletInsertionIterable;
}

private static boolean isMemoryPressureException(final Throwable throwable) {
return LoadTsFileDataTypeConverter.isMemoryPressureException(throwable);
}

@Override
public void close() {
super.close();
Expand Down
Loading
Loading