Skip to content

Add pipe data loss debug logs and split test #18120

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 5 commits into
base: master
Choose a base branch
from
Open

Add pipe data loss debug logs and split test #18120

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
940bdc2
Add pipe data loss debug logs and split test
Caideyipi Jul 6, 2026
bc3ebbe
Fix query parser debug tsfile path scope
Caideyipi Jul 6, 2026
624a5ef
Strengthen pipe data loss debug counters
Caideyipi Jul 6, 2026
55432cf
Merge branch 'master' of https://github.com/apache/iotdb into pipe-da…
Caideyipi Jul 7, 2026
f513dd6
Add temporary pipe data loss debug logs
Caideyipi Jul 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
84 changes: 84 additions & 0 deletions ...main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.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.exception.runtime.TableLostRuntimeException;
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.event.common.util.PipeDataLossDebugUtil;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedDeleteDataNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedInsertNode;
Expand All @@ -54,6 +55,7 @@
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;

public class DataExecutionVisitor implements PlanVisitor<TSStatus, DataRegion> {
Expand Down Expand Up @@ -120,6 +122,13 @@ public TSStatus visitInsertTablet(final InsertTabletNode node, final DataRegion
node.getTimes()[0],
node.getMeasurements(),
e.getFailingStatus());
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} data execution batch failure for InsertTabletNode, node={}, failingStatus={}",
PipeDataLossDebugUtil.PREFIX,
PipeDataLossDebugUtil.formatInsertTabletNode(node),
PipeDataLossDebugUtil.formatStatuses(Arrays.asList(e.getFailingStatus())));
}
// For each error
TSStatus firstStatus = null;
for (final TSStatus status : e.getFailingStatus()) {
Expand All @@ -146,10 +155,26 @@ public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) {
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
} catch (BatchProcessException e) {
LOGGER.warn(DataNodeMiscMessages.BATCH_FAILURE_INSERT_ROWS);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} data execution batch failure for InsertRowsNode, node={}, failingStatus={}",
PipeDataLossDebugUtil.PREFIX,
PipeDataLossDebugUtil.formatInsertRowsNode(node),
PipeDataLossDebugUtil.formatStatuses(node.getResults().values()));
}
TSStatus firstStatus = null;
// for each error
for (Map.Entry<Integer, TSStatus> failedEntry : node.getResults().entrySet()) {
InsertRowNode insertRowNode = node.getInsertRowNodeList().get(failedEntry.getKey());
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} data execution failed InsertRowsNode entry, failedIndex={}, failedStatus={}, insertRowNode={}, parentNode={}",
PipeDataLossDebugUtil.PREFIX,
failedEntry.getKey(),
PipeDataLossDebugUtil.formatStatus(failedEntry.getValue()),
PipeDataLossDebugUtil.formatInsertRowNode(insertRowNode),
PipeDataLossDebugUtil.formatInsertRowsNode(node));
}
if (firstStatus == null) {
firstStatus = failedEntry.getValue();
}
Expand All @@ -162,6 +187,14 @@ public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) {
failedEntry.getValue());
// Return WRITE_PROCESS_REJECT directly for the consensus retry logic
if (failedEntry.getValue().getCode() == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} data execution returning WRITE_PROCESS_REJECT for InsertRowsNode and clearing results, failedIndex={}, failedStatus={}, nodeBeforeClear={}",
PipeDataLossDebugUtil.PREFIX,
failedEntry.getKey(),
PipeDataLossDebugUtil.formatStatus(failedEntry.getValue()),
PipeDataLossDebugUtil.formatInsertRowsNode(node));
}
node.clearResults();
return failedEntry.getValue();
}
Expand All @@ -184,10 +217,29 @@ public TSStatus visitInsertMultiTablets(InsertMultiTabletsNode node, DataRegion
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
} catch (BatchProcessException e) {
LOGGER.warn(DataNodeMiscMessages.BATCH_FAILURE_INSERT_MULTI_TABLETS);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} data execution batch failure for InsertMultiTabletsNode, node={}, failingStatus={}",
PipeDataLossDebugUtil.PREFIX,
PipeDataLossDebugUtil.formatInsertMultiTabletsNode(node),
PipeDataLossDebugUtil.formatStatuses(node.getResults().values()));
}
TSStatus firstStatus = null;
for (Map.Entry<Integer, TSStatus> failedEntry : node.getResults().entrySet()) {
InsertTabletNode insertTabletNode =
node.getInsertTabletNodeList().get(failedEntry.getKey());
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} data execution failed InsertMultiTabletsNode entry, failedIndex={}, parentIndex={}, failedStatus={}, insertTabletNode={}, parentNode={}",
PipeDataLossDebugUtil.PREFIX,
failedEntry.getKey(),
failedEntry.getKey() < node.getParentInsertTabletNodeIndexList().size()
? node.getParentInsertTabletNodeIndexList().get(failedEntry.getKey())
: null,
PipeDataLossDebugUtil.formatStatus(failedEntry.getValue()),
PipeDataLossDebugUtil.formatInsertTabletNode(insertTabletNode),
PipeDataLossDebugUtil.formatInsertMultiTabletsNode(node));
}
if (firstStatus == null) {
firstStatus = failedEntry.getValue();
}
Expand All @@ -200,6 +252,14 @@ public TSStatus visitInsertMultiTablets(InsertMultiTabletsNode node, DataRegion
failedEntry.getValue());
// Return WRITE_PROCESS_REJECT directly for the consensus retry logic
if (failedEntry.getValue().getCode() == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} data execution returning WRITE_PROCESS_REJECT for InsertMultiTabletsNode and clearing results, failedIndex={}, failedStatus={}, nodeBeforeClear={}",
PipeDataLossDebugUtil.PREFIX,
failedEntry.getKey(),
PipeDataLossDebugUtil.formatStatus(failedEntry.getValue()),
PipeDataLossDebugUtil.formatInsertMultiTabletsNode(node));
}
node.clearResults();
return failedEntry.getValue();
}
Expand All @@ -223,9 +283,25 @@ public TSStatus visitInsertRowsOfOneDevice(
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
} catch (BatchProcessException e) {
LOGGER.warn(DataNodeMiscMessages.BATCH_FAILURE_INSERT_ROWS_ONE_DEVICE);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} data execution batch failure for InsertRowsOfOneDeviceNode, node={}, failingStatus={}",
PipeDataLossDebugUtil.PREFIX,
PipeDataLossDebugUtil.formatInsertRowsOfOneDeviceNode(node),
PipeDataLossDebugUtil.formatStatuses(node.getResults().values()));
}
TSStatus firstStatus = null;
for (Map.Entry<Integer, TSStatus> failedEntry : node.getResults().entrySet()) {
InsertRowNode insertRowNode = node.getInsertRowNodeList().get(failedEntry.getKey());
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} data execution failed InsertRowsOfOneDeviceNode entry, failedIndex={}, failedStatus={}, insertRowNode={}, parentNode={}",
PipeDataLossDebugUtil.PREFIX,
failedEntry.getKey(),
PipeDataLossDebugUtil.formatStatus(failedEntry.getValue()),
PipeDataLossDebugUtil.formatInsertRowNode(insertRowNode),
PipeDataLossDebugUtil.formatInsertRowsOfOneDeviceNode(node));
}
if (firstStatus == null) {
firstStatus = failedEntry.getValue();
}
Expand All @@ -238,6 +314,14 @@ public TSStatus visitInsertRowsOfOneDevice(
failedEntry.getValue());
// Return WRITE_PROCESS_REJECT directly for the consensus retry logic
if (failedEntry.getValue().getCode() == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} data execution returning WRITE_PROCESS_REJECT for InsertRowsOfOneDeviceNode and clearing results, failedIndex={}, failedStatus={}, nodeBeforeClear={}",
PipeDataLossDebugUtil.PREFIX,
failedEntry.getKey(),
PipeDataLossDebugUtil.formatStatus(failedEntry.getValue()),
PipeDataLossDebugUtil.formatInsertRowsOfOneDeviceNode(node));
}
node.clearResults();
return failedEntry.getValue();
}
Expand Down
27 changes: 24 additions & 3 deletions ...node/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.util.PipeDataLossDebugUtil;
import org.apache.iotdb.db.pipe.source.schemaregion.IoTDBSchemaRegionSource;
import org.apache.iotdb.db.pipe.source.schemaregion.PipePlanTreePrivilegeParseVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractDeleteDataNode;
Expand Down Expand Up @@ -167,10 +168,30 @@ public static boolean canSkipParsing4TsFileEvent(final PipeTsFileInsertionEvent
}

private void collectParsedRawTableEvent(final PipeRawTabletInsertionEvent parsedEvent) {
if (!parsedEvent.hasNoNeedParsingAndIsEmpty()) {
hasNoGeneratedEvent = false;
collectEvent(parsedEvent);
if (parsedEvent.hasNoNeedParsingAndIsEmpty()) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} collector dropped empty parsed tablet, {}, regionId={}, tablet={}",
PipeDataLossDebugUtil.PREFIX,
PipeDataLossDebugUtil.formatPipe(
parsedEvent.getPipeName(), parsedEvent.getCreationTime()),
regionId,
parsedEvent.getTabletDebugString());
}
return;
}

hasNoGeneratedEvent = false;
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
"{} collector enqueued parsed tablet, {}, regionId={}, tablet={}",
PipeDataLossDebugUtil.PREFIX,
PipeDataLossDebugUtil.formatPipe(
parsedEvent.getPipeName(), parsedEvent.getCreationTime()),
regionId,
parsedEvent.getTabletDebugString());
}
collectEvent(parsedEvent);
}

private void parseAndCollectEvent(final PipeDeleteDataNodeEvent deleteDataEvent) {
Expand Down
5 changes: 5 additions & 0 deletions ...c/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.apache.iotdb.db.pipe.event.common.tablet.parser.TabletInsertionEventTablePatternParser;
import org.apache.iotdb.db.pipe.event.common.tablet.parser.TabletInsertionEventTreePatternParser;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.util.PipeDataLossDebugUtil;
import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
Expand Down Expand Up @@ -411,6 +412,10 @@ public String getDeviceId() {
return Objects.nonNull(tablet) ? tablet.getDeviceId() : deviceId;
}

public String getTabletDebugString() {
return PipeDataLossDebugUtil.formatTablet(tablet);
}

public EnrichedEvent getSourceEvent() {
return sourceEvent;
}
Expand Down
82 changes: 79 additions & 3 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 @@ -36,6 +36,7 @@
import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils.TabletStringInternPool;
import org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParser;
import org.apache.iotdb.db.pipe.event.common.tsfile.parser.util.ModsOperationUtil;
import org.apache.iotdb.db.pipe.event.common.util.PipeDataLossDebugUtil;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
Expand Down Expand Up @@ -76,6 +77,7 @@ public class TsFileInsertionEventQueryParser extends TsFileInsertionEventParser

private final PipeMemoryBlock allocatedMemoryBlock;
private final TsFileReader tsFileReader;
private final String tsFilePath;

private final Iterator<Map.Entry<IDeviceID, List<String>>> deviceMeasurementsMapIterator;
private final Map<IDeviceID, Boolean> deviceIsAlignedMap;
Expand Down Expand Up @@ -180,6 +182,8 @@ public TsFileInsertionEventQueryParser(
sourceEvent,
isWithMod);

tsFilePath = tsFile.getAbsolutePath();

try {
currentModifications =
isWithMod
Expand Down Expand Up @@ -306,8 +310,34 @@ public TsFileInsertionEventQueryParser(
}

// Filter again to get the final deviceMeasurementsMap that exactly matches the pattern.
deviceMeasurementsMapIterator =
filterDeviceMeasurementsMapByPattern(deviceMeasurementsMap).entrySet().iterator();
final Map<IDeviceID, List<String>> filteredDeviceMeasurementsMap =
filterDeviceMeasurementsMapByPattern(deviceMeasurementsMap);
deviceMeasurementsMapIterator = filteredDeviceMeasurementsMap.entrySet().iterator();
if (LOGGER.isDebugEnabled()) {
final int measurementCount =
filteredDeviceMeasurementsMap.values().stream().mapToInt(List::size).sum();
final long metadataPointCount = countMetadataPoints(filteredDeviceMeasurementsMap);
LOGGER.debug(
"{} query parser selected timeseries, {}, tsFile={}, deviceCount={}, measurementCount={}, metadataPointCount={}, isWithMod={}, pattern={}",
PipeDataLossDebugUtil.PREFIX,
PipeDataLossDebugUtil.formatPipe(pipeName, creationTime),
tsFile.getAbsolutePath(),
filteredDeviceMeasurementsMap.size(),
measurementCount,
metadataPointCount,
isWithMod,
treePattern);
filteredDeviceMeasurementsMap.forEach(
(device, measurements) ->
LOGGER.debug(
"{} query parser selected device, {}, tsFile={}, device={}, measurements={}, metadataStatistics={}",
PipeDataLossDebugUtil.PREFIX,
PipeDataLossDebugUtil.formatPipe(pipeName, creationTime),
tsFile.getAbsolutePath(),
device,
measurements,
formatMeasurementStatistics(device, measurements)));
}

// No longer need this. Help GC.
tsFileSequenceReader.clearCachedDeviceMetadata();
Expand Down Expand Up @@ -341,6 +371,49 @@ public TsFileInsertionEventQueryParser(
isWithMod);
}

private List<String> formatMeasurementStatistics(
final IDeviceID deviceId, final List<String> measurements) {
final List<String> result = new ArrayList<>(measurements.size());
for (final String measurement : measurements) {
try {
final TimeseriesMetadata metadata =
tsFileSequenceReader.readTimeseriesMetadata(deviceId, measurement, true);
result.add(
measurement
+ (metadata == null
? "{metadata=null}"
: "{count="
+ metadata.getStatistics().getCount()
+ ", startTime="
+ metadata.getStatistics().getStartTime()
+ ", endTime="
+ metadata.getStatistics().getEndTime()
+ "}"));
} catch (final Exception e) {
result.add(measurement + "{failedToReadStatistics=" + e.getMessage() + "}");
}
}
return result;
}

private long countMetadataPoints(final Map<IDeviceID, List<String>> deviceMeasurementsMap) {
long result = 0;
for (final Map.Entry<IDeviceID, List<String>> entry : deviceMeasurementsMap.entrySet()) {
for (final String measurement : entry.getValue()) {
try {
final TimeseriesMetadata metadata =
tsFileSequenceReader.readTimeseriesMetadata(entry.getKey(), measurement, true);
if (Objects.nonNull(metadata)) {
result += metadata.getStatistics().getCount();
}
} catch (final Exception ignored) {
// The per-measurement metadata debug log records the failure details.
}
}
}
return result;
}

private Map<IDeviceID, List<String>> filterDeviceMeasurementsMapByPattern(
final Map<IDeviceID, List<String>> originalDeviceMeasurementsMap)
throws IllegalPathException {
Expand Down Expand Up @@ -497,7 +570,10 @@ public boolean hasNext() {
timeFilterExpression,
allocatedMemoryBlockForTablet,
currentModifications,
tabletStringInternPool);
tabletStringInternPool,
pipeName,
creationTime,
tsFilePath);
} catch (final Exception e) {
close();
throw new PipeException(
Expand Down
Loading
Loading