diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java index bd48ed25ae759..377705f695e88 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java @@ -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; @@ -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 { @@ -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()) { @@ -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 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(); } @@ -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(); } @@ -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 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(); } @@ -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(); } @@ -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 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(); } @@ -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(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java index 2324a3d27076f..af58a2bfc804c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java @@ -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; @@ -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) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java index 5bdef252a2f4a..7825323beccbd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java @@ -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; @@ -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; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java index ba840b11c32dc..6e60dc5ae5da2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java @@ -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; @@ -76,6 +77,7 @@ public class TsFileInsertionEventQueryParser extends TsFileInsertionEventParser private final PipeMemoryBlock allocatedMemoryBlock; private final TsFileReader tsFileReader; + private final String tsFilePath; private final Iterator>> deviceMeasurementsMapIterator; private final Map deviceIsAlignedMap; @@ -180,6 +182,8 @@ public TsFileInsertionEventQueryParser( sourceEvent, isWithMod); + tsFilePath = tsFile.getAbsolutePath(); + try { currentModifications = isWithMod @@ -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> 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(); @@ -341,6 +371,49 @@ public TsFileInsertionEventQueryParser( isWithMod); } + private List formatMeasurementStatistics( + final IDeviceID deviceId, final List measurements) { + final List 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> deviceMeasurementsMap) { + long result = 0; + for (final Map.Entry> 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> filterDeviceMeasurementsMapByPattern( final Map> originalDeviceMeasurementsMap) throws IllegalPathException { @@ -497,7 +570,10 @@ public boolean hasNext() { timeFilterExpression, allocatedMemoryBlockForTablet, currentModifications, - tabletStringInternPool); + tabletStringInternPool, + pipeName, + creationTime, + tsFilePath); } catch (final Exception e) { close(); throw new PipeException( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParserTabletIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParserTabletIterator.java index 5c7c06089fe9b..6bb1ecc1d558f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParserTabletIterator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParserTabletIterator.java @@ -24,6 +24,7 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils; import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils.TabletStringInternPool; 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; @@ -46,6 +47,8 @@ import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.ArrayList; @@ -58,6 +61,9 @@ public class TsFileInsertionEventQueryParserTabletIterator implements Iterator { + private static final Logger LOGGER = + LoggerFactory.getLogger(TsFileInsertionEventQueryParserTabletIterator.class); + private final TsFileReader tsFileReader; private final Map measurementDataTypeMap; @@ -77,6 +83,10 @@ public class TsFileInsertionEventQueryParserTabletIterator implements Iterator measurementDataTypeMap, @@ -85,13 +95,19 @@ public class TsFileInsertionEventQueryParserTabletIterator implements Iterator currentModifications, - final TabletStringInternPool tabletStringInternPool) + final TabletStringInternPool tabletStringInternPool, + final String pipeName, + final long creationTime, + final String tsFilePath) throws IOException { this.tsFileReader = tsFileReader; this.measurementDataTypeMap = measurementDataTypeMap; this.deviceId = deviceId; this.deviceIdString = tabletStringInternPool.intern(deviceId.toString()); + this.pipeName = pipeName; + this.creationTime = creationTime; + this.tsFilePath = tsFilePath; this.measurements = measurements.stream() .filter( @@ -159,9 +175,25 @@ private Tablet buildNextTablet() throws IOException { return tablet; } + int inputRowCount = 0; + int droppedAllNullRowCount = 0; + int nullFieldCount = 0; + int deletedFieldCount = 0; + int nonNullFieldCount = 0; + long firstInputTime = Long.MIN_VALUE; + long lastInputTime = Long.MIN_VALUE; + long firstOutputTime = Long.MIN_VALUE; + long lastOutputTime = Long.MIN_VALUE; + boolean isFirstRow = true; while (queryDataSet.hasNext()) { final RowRecord rowRecord = this.rowRecord != null ? this.rowRecord : queryDataSet.next(); + final long timestamp = rowRecord.getTimestamp(); + ++inputRowCount; + if (firstInputTime == Long.MIN_VALUE) { + firstInputTime = timestamp; + } + lastInputTime = timestamp; if (isFirstRow) { // Calculate row count and memory size of the tablet based on the first row this.rowRecord = rowRecord; // Save the first row for later use @@ -188,21 +220,34 @@ private Tablet buildNextTablet() throws IOException { final Field field = fields.get(i); final String measurement = measurements.get(i); final TSDataType dataType = schemas.get(i).getType(); + final boolean isDeleted = + field != null && ModsOperationUtil.isDelete(timestamp, measurementModsList.get(i)); // Check if this value is deleted by mods - if (field == null - || ModsOperationUtil.isDelete(rowRecord.getTimestamp(), measurementModsList.get(i))) { + if (field == null || isDeleted) { + if (field == null) { + ++nullFieldCount; + } else { + ++deletedFieldCount; + } if (dataType.isBinary()) { PipeTabletUtils.putValue(tablet, rowIndex, i, dataType, Binary.EMPTY_VALUE); } PipeTabletUtils.markNullValue(tablet, rowIndex, i); } else { + ++nonNullFieldCount; PipeTabletUtils.putValue( tablet, rowIndex, i, dataType, field.getObjectValue(schemas.get(i).getType())); isNeedFillTime = true; } } if (isNeedFillTime) { - PipeTabletUtils.putTimestamp(tablet, rowIndex, rowRecord.getTimestamp()); + if (firstOutputTime == Long.MIN_VALUE) { + firstOutputTime = timestamp; + } + lastOutputTime = timestamp; + PipeTabletUtils.putTimestamp(tablet, rowIndex, timestamp); + } else { + ++droppedAllNullRowCount; } if (tablet.getRowSize() == tablet.getMaxRowNumber()) { @@ -211,6 +256,26 @@ private Tablet buildNextTablet() throws IOException { } PipeTabletUtils.compactBitMaps(tablet); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} query parser emitted tablet, {}, tsFile={}, device={}, measurements={}, inputRows={}, outputRows={}, droppedAllNullRows={}, nullFields={}, deletedFields={}, nonNullFields={}, firstInputTime={}, lastInputTime={}, firstOutputTime={}, lastOutputTime={}, tablet={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatPipe(pipeName, creationTime), + tsFilePath, + deviceIdString, + measurements, + inputRowCount, + tablet.getRowSize(), + droppedAllNullRowCount, + nullFieldCount, + deletedFieldCount, + nonNullFieldCount, + firstInputTime == Long.MIN_VALUE ? "null" : String.valueOf(firstInputTime), + lastInputTime == Long.MIN_VALUE ? "null" : String.valueOf(lastInputTime), + firstOutputTime == Long.MIN_VALUE ? "null" : String.valueOf(firstOutputTime), + lastOutputTime == Long.MIN_VALUE ? "null" : String.valueOf(lastOutputTime), + PipeDataLossDebugUtil.formatTablet(tablet)); + } return tablet; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/util/PipeDataLossDebugUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/util/PipeDataLossDebugUtil.java new file mode 100644 index 0000000000000..5e9b98b032661 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/util/PipeDataLossDebugUtil.java @@ -0,0 +1,903 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.event.common.util; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.statement.Statement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertMultiTabletsStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; +import org.apache.iotdb.pipe.api.event.Event; +import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; +import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; + +import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; + +import java.io.File; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; + +public final class PipeDataLossDebugUtil { + + public static final String PREFIX = "[PipeDataLossDebug]"; + + private static final int MAX_PRINTED_MEASUREMENTS = 16; + private static final int MAX_PRINTED_STATEMENTS = 16; + private static final int MAX_PRINTED_STATUSES = 16; + private static final int MAX_PRINTED_EVENTS = 32; + private static final int MAX_PRINTED_PLAN_NODES = 16; + + private PipeDataLossDebugUtil() {} + + public static String formatPipe(final String pipeName, final long creationTime) { + return "pipeName=" + pipeName + ", creationTime=" + creationTime; + } + + public static String formatReq(final TPipeTransferReq req) { + if (Objects.isNull(req)) { + return "req=null"; + } + + final byte[] body = req.getBody(); + return "version=" + + req.getVersion() + + ", type=" + + req.getType() + + ", bodySize=" + + (Objects.isNull(body) ? "null" : body.length) + + ", bodyHash=" + + (Objects.isNull(body) ? "null" : Arrays.hashCode(body)); + } + + public static String formatException(final Exception exception) { + if (Objects.isNull(exception)) { + return "exception=null"; + } + return "exceptionClass=" + + exception.getClass().getName() + + ", message=" + + exception.getMessage(); + } + + public static String formatEvent(final Event event) { + if (Objects.isNull(event)) { + return "event=null"; + } + + final StringBuilder builder = + new StringBuilder("eventClass=") + .append(event.getClass().getSimpleName()) + .append(", identity=") + .append(System.identityHashCode(event)); + + if (event instanceof EnrichedEvent) { + final EnrichedEvent enrichedEvent = (EnrichedEvent) event; + builder + .append(", ") + .append(formatPipe(enrichedEvent.getPipeName(), enrichedEvent.getCreationTime())) + .append(", regionId=") + .append(enrichedEvent.getRegionId()) + .append(", committerKey=") + .append(enrichedEvent.getCommitterKey()) + .append(", commitId=") + .append(enrichedEvent.getCommitId()) + .append(", commitIds=") + .append(enrichedEvent.getCommitIds()) + .append(", referenceCount=") + .append(enrichedEvent.getReferenceCount()) + .append(", released=") + .append(enrichedEvent.isReleased()) + .append(", shouldReportOnCommit=") + .append(enrichedEvent.isShouldReportOnCommit()) + .append(", progressIndex=") + .append(enrichedEvent.getProgressIndex()) + .append(", eventTimeRange=[") + .append(enrichedEvent.getStartTime()) + .append(",") + .append(enrichedEvent.getEndTime()) + .append("]") + .append(", generatedByPipe=") + .append(safeIsGeneratedByPipe(enrichedEvent)); + } + + if (event instanceof PipeRawTabletInsertionEvent) { + final PipeRawTabletInsertionEvent rawEvent = (PipeRawTabletInsertionEvent) event; + builder + .append(", device=") + .append(rawEvent.getDeviceId()) + .append(", aligned=") + .append(rawEvent.isAligned()) + .append(", tablet={") + .append(rawEvent.getTabletDebugString()) + .append("}"); + } else if (event instanceof PipeInsertNodeTabletInsertionEvent) { + final PipeInsertNodeTabletInsertionEvent insertNodeEvent = + (PipeInsertNodeTabletInsertionEvent) event; + builder + .append(", device=") + .append(insertNodeEvent.getDeviceId()) + .append(", insertNode={") + .append(formatInsertNode(insertNodeEvent.getInsertNode())) + .append("}"); + } else if (event instanceof PipeTsFileInsertionEvent) { + final PipeTsFileInsertionEvent tsFileEvent = (PipeTsFileInsertionEvent) event; + final File tsFile = tsFileEvent.getTsFile(); + builder + .append(", tsFile=") + .append(Objects.isNull(tsFile) ? null : tsFile.getAbsolutePath()) + .append(", tsFileLength=") + .append(Objects.isNull(tsFile) ? null : tsFile.length()); + } + + return builder.toString(); + } + + public static String formatEvents(final Iterable events) { + if (Objects.isNull(events)) { + return "events=null"; + } + + final List samples = new ArrayList<>(); + final Map pipeCounters = new LinkedHashMap<>(); + int eventCount = 0; + for (final Event event : events) { + ++eventCount; + if (event instanceof EnrichedEvent) { + final EnrichedEvent enrichedEvent = (EnrichedEvent) event; + final String pipeKey = + enrichedEvent.getPipeName() + + "@" + + enrichedEvent.getCreationTime() + + "#" + + enrichedEvent.getRegionId(); + pipeCounters.compute(pipeKey, (key, value) -> Objects.isNull(value) ? 1 : value + 1); + } + if (samples.size() < MAX_PRINTED_EVENTS) { + samples.add(formatEvent(event)); + } + } + + return "eventCount=" + + eventCount + + ", pipeCounters=" + + pipeCounters + + ", samples=" + + samples + + (eventCount > MAX_PRINTED_EVENTS ? "...(" + eventCount + ")" : ""); + } + + public static String formatTablet(final Tablet tablet) { + if (Objects.isNull(tablet)) { + return "tablet=null"; + } + + final int rowSize = tablet.getRowSize(); + final List schemas = tablet.getSchemas(); + final int schemaSize = Objects.isNull(schemas) ? 0 : schemas.size(); + final long markedNullCells = countMarkedCells(tablet.getBitMaps(), rowSize); + final long[] timestamps = tablet.getTimestamps(); + final String firstTime = + rowSize > 0 && Objects.nonNull(timestamps) && timestamps.length > 0 + ? String.valueOf(timestamps[0]) + : "null"; + final String lastTime = + rowSize > 0 && Objects.nonNull(timestamps) && timestamps.length >= rowSize + ? String.valueOf(timestamps[rowSize - 1]) + : "null"; + + return "device=" + + safeGetDeviceId(tablet) + + ", rowSize=" + + rowSize + + ", schemaSize=" + + schemaSize + + ", dataPointCount=" + + countDataPoints(rowSize, schemaSize, markedNullCells) + + ", measurements=" + + formatMeasurements(schemas) + + ", firstTime=" + + firstTime + + ", lastTime=" + + lastTime + + ", markedNullCells=" + + markedNullCells; + } + + public static String formatTabletInsertionEvent(final TabletInsertionEvent event) { + if (Objects.isNull(event)) { + return "tabletInsertionEvent=null"; + } + if (event instanceof PipeRawTabletInsertionEvent) { + final PipeRawTabletInsertionEvent rawEvent = (PipeRawTabletInsertionEvent) event; + return "rawTablet, aligned=" + + rawEvent.isAligned() + + ", device=" + + rawEvent.getDeviceId() + + ", " + + rawEvent.getTabletDebugString(); + } + if (event instanceof PipeInsertNodeTabletInsertionEvent) { + final PipeInsertNodeTabletInsertionEvent insertNodeEvent = + (PipeInsertNodeTabletInsertionEvent) event; + return "insertNodeTablet, device=" + + insertNodeEvent.getDeviceId() + + ", " + + formatInsertNode(insertNodeEvent.getInsertNode()); + } + return String.valueOf(event); + } + + public static String formatInsertNode(final InsertNode node) { + if (Objects.isNull(node)) { + return "insertNode=null"; + } + if (node instanceof InsertTabletNode) { + return formatInsertTabletNode((InsertTabletNode) node); + } + if (node instanceof InsertRowsNode) { + return formatInsertRowsNode((InsertRowsNode) node); + } + if (node instanceof InsertMultiTabletsNode) { + return formatInsertMultiTabletsNode((InsertMultiTabletsNode) node); + } + if (node instanceof InsertRowNode) { + return formatInsertRowNode((InsertRowNode) node); + } + return "type=" + + node.getType() + + ", class=" + + node.getClass().getName() + + ", device=" + + node.getTargetPath() + + ", aligned=" + + node.isAligned() + + ", generatedByPipe=" + + node.isGeneratedByPipe() + + ", progressIndex=" + + node.getProgressIndex() + + ", measurements=" + + formatMeasurements(node.getMeasurements()); + } + + public static String formatInsertTabletNode(final InsertTabletNode node) { + if (Objects.isNull(node)) { + return "insertTabletNode=null"; + } + + final int rowCount = node.getRowCount(); + final int measurementCount = measurementCount(node.getMeasurements()); + final long markedNullCells = countMarkedCells(node.getBitMaps(), rowCount); + return "type=" + + node.getType() + + ", device=" + + node.getTargetPath() + + ", aligned=" + + node.isAligned() + + ", generatedByPipe=" + + node.isGeneratedByPipe() + + ", progressIndex=" + + node.getProgressIndex() + + ", rowCount=" + + rowCount + + ", measurementCount=" + + measurementCount + + ", dataPointCount=" + + countDataPoints(rowCount, measurementCount, markedNullCells) + + ", measurements=" + + formatMeasurements(node.getMeasurements()) + + ", firstTime=" + + firstTime(node.getTimes(), rowCount) + + ", lastTime=" + + lastTime(node.getTimes(), rowCount) + + ", markedNullCells=" + + markedNullCells; + } + + public static String formatInsertRowNode(final InsertRowNode node) { + if (Objects.isNull(node)) { + return "insertRowNode=null"; + } + + return "type=" + + node.getType() + + ", device=" + + node.getTargetPath() + + ", aligned=" + + node.isAligned() + + ", generatedByPipe=" + + node.isGeneratedByPipe() + + ", progressIndex=" + + node.getProgressIndex() + + ", time=" + + node.getTime() + + ", measurementCount=" + + measurementCount(node.getMeasurements()) + + ", measurements=" + + formatMeasurements(node.getMeasurements()); + } + + public static String formatInsertRowsNode(final InsertRowsNode node) { + if (Objects.isNull(node)) { + return "insertRowsNode=null"; + } + + final List insertRowNodes = node.getInsertRowNodeList(); + return "type=" + + node.getType() + + ", generatedByPipe=" + + node.isGeneratedByPipe() + + ", progressIndex=" + + node.getProgressIndex() + + ", rowNodeCount=" + + insertRowNodes.size() + + ", indexes=" + + node.getInsertRowNodeIndexList() + + ", results=" + + formatIndexedStatuses(node.getResults()) + + ", samples=" + + insertRowNodes.stream() + .limit(MAX_PRINTED_PLAN_NODES) + .map(PipeDataLossDebugUtil::formatInsertRowNode) + .collect(Collectors.toList()) + + (insertRowNodes.size() > MAX_PRINTED_PLAN_NODES + ? "...(" + insertRowNodes.size() + ")" + : ""); + } + + public static String formatInsertRowsOfOneDeviceNode(final InsertRowsOfOneDeviceNode node) { + if (Objects.isNull(node)) { + return "insertRowsOfOneDeviceNode=null"; + } + + final List insertRowNodes = node.getInsertRowNodeList(); + return "type=" + + node.getType() + + ", device=" + + node.getTargetPath() + + ", aligned=" + + node.isAligned() + + ", generatedByPipe=" + + node.isGeneratedByPipe() + + ", progressIndex=" + + node.getProgressIndex() + + ", rowNodeCount=" + + insertRowNodes.size() + + ", indexes=" + + node.getInsertRowNodeIndexList() + + ", results=" + + formatIndexedStatuses(node.getResults()) + + ", samples=" + + insertRowNodes.stream() + .limit(MAX_PRINTED_PLAN_NODES) + .map(PipeDataLossDebugUtil::formatInsertRowNode) + .collect(Collectors.toList()) + + (insertRowNodes.size() > MAX_PRINTED_PLAN_NODES + ? "...(" + insertRowNodes.size() + ")" + : ""); + } + + public static String formatInsertMultiTabletsNode(final InsertMultiTabletsNode node) { + if (Objects.isNull(node)) { + return "insertMultiTabletsNode=null"; + } + + final List insertTabletNodes = node.getInsertTabletNodeList(); + return "type=" + + node.getType() + + ", generatedByPipe=" + + node.isGeneratedByPipe() + + ", progressIndex=" + + node.getProgressIndex() + + ", tabletNodeCount=" + + insertTabletNodes.size() + + ", parentIndexes=" + + node.getParentInsertTabletNodeIndexList() + + ", results=" + + formatIndexedStatuses(node.getResults()) + + ", totalRows=" + + insertTabletNodes.stream().mapToLong(InsertTabletNode::getRowCount).sum() + + ", totalDataPointCount=" + + insertTabletNodes.stream().mapToLong(PipeDataLossDebugUtil::countDataPoints).sum() + + ", samples=" + + insertTabletNodes.stream() + .limit(MAX_PRINTED_PLAN_NODES) + .map(PipeDataLossDebugUtil::formatInsertTabletNode) + .collect(Collectors.toList()) + + (insertTabletNodes.size() > MAX_PRINTED_PLAN_NODES + ? "...(" + insertTabletNodes.size() + ")" + : ""); + } + + public static String formatStatement(final Statement statement) { + if (Objects.isNull(statement)) { + return "statement=null"; + } + + if (statement instanceof InsertTabletStatement) { + final InsertTabletStatement insertTabletStatement = (InsertTabletStatement) statement; + final int rowCount = insertTabletStatement.getRowCount(); + final int measurementCount = measurementCount(insertTabletStatement.getMeasurements()); + final long markedNullCells = + countMarkedCells(insertTabletStatement.getBitMaps(), insertTabletStatement.getRowCount()); + return "type=" + + statement.getType() + + ", database=" + + insertTabletStatement.getDatabaseName().orElse(null) + + ", writeToTable=" + + insertTabletStatement.isWriteToTable() + + ", device=" + + insertTabletStatement.getDevicePath() + + ", rowCount=" + + rowCount + + ", measurementCount=" + + measurementCount + + ", dataPointCount=" + + countDataPoints(rowCount, measurementCount, markedNullCells) + + ", measurements=" + + formatMeasurements(insertTabletStatement.getMeasurements()) + + ", firstTime=" + + firstTime(insertTabletStatement.getTimes(), rowCount) + + ", lastTime=" + + lastTime(insertTabletStatement.getTimes(), rowCount) + + ", markedNullCells=" + + markedNullCells + + ", failedMeasurements=" + + insertTabletStatement.getFailedMeasurements(); + } + + if (statement instanceof InsertMultiTabletsStatement) { + final InsertMultiTabletsStatement insertMultiTabletsStatement = + (InsertMultiTabletsStatement) statement; + final List statements = + insertMultiTabletsStatement.getInsertTabletStatementList(); + return "type=" + + statement.getType() + + ", database=" + + insertMultiTabletsStatement.getDatabaseName().orElse(null) + + ", tabletCount=" + + statements.size() + + ", totalRows=" + + statements.stream().mapToInt(InsertTabletStatement::getRowCount).sum() + + ", totalDataPointCount=" + + statements.stream().mapToLong(PipeDataLossDebugUtil::countDataPoints).sum() + + ", totalMarkedNullCells=" + + statements.stream().mapToLong(PipeDataLossDebugUtil::countMarkedCells).sum() + + ", samples=" + + statements.stream() + .limit(MAX_PRINTED_STATEMENTS) + .map(PipeDataLossDebugUtil::formatStatement) + .collect(Collectors.toList()) + + (statements.size() > MAX_PRINTED_STATEMENTS ? "...(" + statements.size() + ")" : ""); + } + + if (statement instanceof InsertRowsStatement) { + final InsertRowsStatement insertRowsStatement = (InsertRowsStatement) statement; + return "type=" + + statement.getType() + + ", database=" + + insertRowsStatement.getDatabaseName().orElse(null) + + ", rowStatementCount=" + + insertRowsStatement.getInsertRowStatementList().size() + + ", totalDataPointCount=" + + insertRowsStatement.getInsertRowStatementList().stream() + .mapToLong(PipeDataLossDebugUtil::countDataPoints) + .sum() + + ", samples=" + + insertRowsStatement.getInsertRowStatementList().stream() + .limit(MAX_PRINTED_STATEMENTS) + .map(PipeDataLossDebugUtil::formatStatement) + .collect(Collectors.toList()) + + (insertRowsStatement.getInsertRowStatementList().size() > MAX_PRINTED_STATEMENTS + ? "...(" + insertRowsStatement.getInsertRowStatementList().size() + ")" + : ""); + } + + if (statement instanceof InsertRowStatement) { + final InsertRowStatement insertRowStatement = (InsertRowStatement) statement; + return "type=" + + statement.getType() + + ", database=" + + insertRowStatement.getDatabaseName().orElse(null) + + ", device=" + + insertRowStatement.getDevicePath() + + ", time=" + + insertRowStatement.getTime() + + ", measurementCount=" + + measurementCount(insertRowStatement.getMeasurements()) + + ", dataPointCount=" + + countDataPoints(insertRowStatement) + + ", measurements=" + + formatMeasurements(insertRowStatement.getMeasurements()) + + ", failedMeasurements=" + + insertRowStatement.getFailedMeasurements(); + } + + return "type=" + statement.getType() + ", class=" + statement.getClass().getName(); + } + + public static String formatStatements(final Collection statements) { + if (Objects.isNull(statements)) { + return "statements=null"; + } + + return "statementCount=" + + statements.size() + + ", totalRows=" + + statements.stream().mapToLong(PipeDataLossDebugUtil::countRows).sum() + + ", totalDataPointCount=" + + statements.stream().mapToLong(PipeDataLossDebugUtil::countDataPoints).sum() + + ", totalMarkedNullCells=" + + statements.stream().mapToLong(PipeDataLossDebugUtil::countMarkedCells).sum() + + ", samples=" + + statements.stream() + .limit(MAX_PRINTED_STATEMENTS) + .map(PipeDataLossDebugUtil::formatStatement) + .collect(Collectors.toList()) + + (statements.size() > MAX_PRINTED_STATEMENTS ? "...(" + statements.size() + ")" : ""); + } + + public static String formatStatus(final TSStatus status) { + if (Objects.isNull(status)) { + return "status=null"; + } + return formatSingleStatus(status) + + (status.isSetSubStatus() + ? ", subStatuses=" + formatStatusesWithIndexes(status.getSubStatus()) + : ""); + } + + public static String formatStatusHeader(final TSStatus status) { + if (Objects.isNull(status)) { + return "status=null"; + } + return "code=" + + status.getCode() + + ", subStatusSize=" + + status.getSubStatusSize() + + ", message=" + + status.getMessage(); + } + + public static String formatStatuses(final Collection statuses) { + if (Objects.isNull(statuses)) { + return "statuses=null"; + } + + return statuses.stream() + .limit(MAX_PRINTED_STATUSES) + .map(PipeDataLossDebugUtil::formatStatus) + .collect(Collectors.toList()) + + (statuses.size() > MAX_PRINTED_STATUSES ? "...(" + statuses.size() + ")" : ""); + } + + public static String formatStatementStatusMapping( + final Statement statement, final TSStatus status) { + if (Objects.isNull(statement) || Objects.isNull(status) || !status.isSetSubStatus()) { + return "statementStatusMapping=none"; + } + + if (statement instanceof InsertMultiTabletsStatement) { + final List statements = + ((InsertMultiTabletsStatement) statement).getInsertTabletStatementList(); + return formatIndexedStatementStatusMapping(statements, status.getSubStatus()); + } + + if (statement instanceof InsertRowsStatement) { + final List statements = + ((InsertRowsStatement) statement).getInsertRowStatementList(); + return formatIndexedStatementStatusMapping(statements, status.getSubStatus()); + } + + return "statementStatusMapping=unsupported, statementType=" + statement.getType(); + } + + public static String formatQueueState( + final int retryEventQueueSize, + final int retryTsFileQueueSize, + final int tabletEventCount, + final int tsFileEventCount) { + return "retryEventQueueSize=" + + retryEventQueueSize + + ", retryTsFileQueueSize=" + + retryTsFileQueueSize + + ", countedTabletEvents=" + + tabletEventCount + + ", countedTsFileEvents=" + + tsFileEventCount; + } + + public static long countDataPoints(final Tablet tablet) { + if (Objects.isNull(tablet)) { + return 0; + } + return countDataPoints( + tablet.getRowSize(), + Objects.isNull(tablet.getSchemas()) ? 0 : tablet.getSchemas().size(), + countMarkedCells(tablet.getBitMaps(), tablet.getRowSize())); + } + + public static long countMarkedCells(final BitMap[] bitMaps, final int rowSize) { + if (Objects.isNull(bitMaps) || rowSize <= 0) { + return 0; + } + + long count = 0; + for (final BitMap bitMap : bitMaps) { + if (Objects.isNull(bitMap)) { + continue; + } + for (int i = 0; i < rowSize; ++i) { + if (bitMap.isMarked(i)) { + ++count; + } + } + } + return count; + } + + private static String formatSingleStatus(final TSStatus status) { + return "code=" + + status.getCode() + + ", message=" + + status.getMessage() + + (status.isSetRedirectNode() ? ", redirectNode=" + status.getRedirectNode() : ""); + } + + private static String formatStatusesWithIndexes(final List statuses) { + if (Objects.isNull(statuses)) { + return "null"; + } + + final List samples = new ArrayList<>(); + final List failures = new ArrayList<>(); + for (int i = 0; i < statuses.size(); ++i) { + final TSStatus status = statuses.get(i); + final String formatted = "index=" + i + "{" + formatSingleStatus(status) + "}"; + if (samples.size() < MAX_PRINTED_STATUSES) { + samples.add(formatted); + } + if (Objects.nonNull(status) + && status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() + && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode() + && failures.size() < MAX_PRINTED_STATUSES) { + failures.add(formatted); + } + } + + return "size=" + + statuses.size() + + ", samples=" + + samples + + (statuses.size() > MAX_PRINTED_STATUSES ? "...(" + statuses.size() + ")" : "") + + ", failures=" + + failures; + } + + private static String formatIndexedStatuses(final Map indexedStatuses) { + if (Objects.isNull(indexedStatuses)) { + return "null"; + } + + return indexedStatuses.entrySet().stream() + .limit(MAX_PRINTED_STATUSES) + .map(entry -> "index=" + entry.getKey() + "{" + formatStatus(entry.getValue()) + "}") + .collect(Collectors.toList()) + + (indexedStatuses.size() > MAX_PRINTED_STATUSES + ? "...(" + indexedStatuses.size() + ")" + : ""); + } + + private static String formatIndexedStatementStatusMapping( + final List statements, final List statuses) { + final int size = Math.max(statements.size(), statuses.size()); + final List samples = new ArrayList<>(); + final List failures = new ArrayList<>(); + + for (int i = 0; i < size; ++i) { + final Statement statement = i < statements.size() ? statements.get(i) : null; + final TSStatus status = i < statuses.size() ? statuses.get(i) : null; + final String formatted = + "index=" + + i + + ", status={" + + formatStatus(status) + + "}, statement={" + + formatStatement(statement) + + "}"; + if (samples.size() < MAX_PRINTED_STATEMENTS) { + samples.add(formatted); + } + if (Objects.nonNull(status) + && status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() + && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode() + && failures.size() < MAX_PRINTED_STATEMENTS) { + failures.add(formatted); + } + } + + return "mappingSize=" + + size + + ", samples=" + + samples + + (size > MAX_PRINTED_STATEMENTS ? "...(" + size + ")" : "") + + ", failures=" + + failures; + } + + private static long countRows(final Statement statement) { + if (statement instanceof InsertTabletStatement) { + return ((InsertTabletStatement) statement).getRowCount(); + } + if (statement instanceof InsertMultiTabletsStatement) { + return ((InsertMultiTabletsStatement) statement) + .getInsertTabletStatementList().stream() + .mapToLong(InsertTabletStatement::getRowCount) + .sum(); + } + if (statement instanceof InsertRowsStatement) { + return ((InsertRowsStatement) statement).getInsertRowStatementList().size(); + } + if (statement instanceof InsertRowStatement) { + return 1; + } + return 0; + } + + private static long countDataPoints(final Statement statement) { + if (statement instanceof InsertTabletStatement) { + return countDataPoints((InsertTabletStatement) statement); + } + if (statement instanceof InsertMultiTabletsStatement) { + return ((InsertMultiTabletsStatement) statement) + .getInsertTabletStatementList().stream() + .mapToLong(PipeDataLossDebugUtil::countDataPoints) + .sum(); + } + if (statement instanceof InsertRowsStatement) { + return ((InsertRowsStatement) statement) + .getInsertRowStatementList().stream() + .mapToLong(PipeDataLossDebugUtil::countDataPoints) + .sum(); + } + if (statement instanceof InsertRowStatement) { + return countDataPoints((InsertRowStatement) statement); + } + return 0; + } + + private static long countDataPoints(final InsertTabletStatement statement) { + final int rowCount = statement.getRowCount(); + return countDataPoints( + rowCount, + measurementCount(statement.getMeasurements()), + countMarkedCells(statement.getBitMaps(), rowCount)); + } + + private static long countDataPoints(final InsertTabletNode node) { + if (Objects.isNull(node)) { + return 0; + } + return countDataPoints( + node.getRowCount(), + measurementCount(node.getMeasurements()), + countMarkedCells(node.getBitMaps(), node.getRowCount())); + } + + private static long countDataPoints(final InsertRowStatement statement) { + return Math.max(0, measurementCount(statement.getMeasurements())); + } + + private static long countDataPoints( + final int rowCount, final int measurementCount, final long markedNullCells) { + return Math.max(0, (long) rowCount * measurementCount - markedNullCells); + } + + private static long countMarkedCells(final Statement statement) { + if (statement instanceof InsertTabletStatement) { + final InsertTabletStatement insertTabletStatement = (InsertTabletStatement) statement; + return countMarkedCells( + insertTabletStatement.getBitMaps(), insertTabletStatement.getRowCount()); + } + if (statement instanceof InsertMultiTabletsStatement) { + return ((InsertMultiTabletsStatement) statement) + .getInsertTabletStatementList().stream() + .mapToLong(PipeDataLossDebugUtil::countMarkedCells) + .sum(); + } + return 0; + } + + private static long countMarkedCells(final InsertTabletStatement statement) { + return countMarkedCells(statement.getBitMaps(), statement.getRowCount()); + } + + private static String safeGetDeviceId(final Tablet tablet) { + try { + return tablet.getDeviceId(); + } catch (final Exception e) { + return "unknown"; + } + } + + private static boolean safeIsGeneratedByPipe(final EnrichedEvent event) { + try { + return event.isGeneratedByPipe(); + } catch (final Exception e) { + return false; + } + } + + private static int measurementCount(final String[] measurements) { + return Objects.isNull(measurements) ? 0 : measurements.length; + } + + private static String formatMeasurements(final List schemas) { + if (Objects.isNull(schemas)) { + return "null"; + } + return schemas.stream() + .limit(MAX_PRINTED_MEASUREMENTS) + .map(schema -> Objects.isNull(schema) ? "null" : schema.getMeasurementName()) + .collect(Collectors.toList()) + + (schemas.size() > MAX_PRINTED_MEASUREMENTS ? "...(" + schemas.size() + ")" : ""); + } + + private static String formatMeasurements(final String[] measurements) { + if (Objects.isNull(measurements)) { + return "null"; + } + return Arrays.stream(measurements).limit(MAX_PRINTED_MEASUREMENTS).collect(Collectors.toList()) + + (measurements.length > MAX_PRINTED_MEASUREMENTS + ? "...(" + measurements.length + ")" + : ""); + } + + private static String firstTime(final long[] times, final int rowCount) { + return rowCount > 0 && Objects.nonNull(times) && times.length > 0 + ? String.valueOf(times[0]) + : "null"; + } + + private static String lastTime(final long[] times, final int rowCount) { + return rowCount > 0 && Objects.nonNull(times) && times.length >= rowCount + ? String.valueOf(times[rowCount - 1]) + : "null"; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java index f4a4baed69495..3d0c40cd72180 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java @@ -55,6 +55,7 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionSnapshotEvent; +import org.apache.iotdb.db.pipe.event.common.util.PipeDataLossDebugUtil; import org.apache.iotdb.db.pipe.metric.receiver.PipeDataNodeReceiverMetrics; import org.apache.iotdb.db.pipe.receiver.visitor.PipePlanToStatementVisitor; import org.apache.iotdb.db.pipe.receiver.visitor.PipeStatementExceptionVisitor; @@ -219,6 +220,14 @@ public synchronized TPipeTransferResp receive(final TPipeTransferReq req) { final short rawRequestType = req.getType(); if (PipeRequestType.isValidatedRequestType(rawRequestType)) { final PipeRequestType requestType = PipeRequestType.valueOf(rawRequestType); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver received request, receiverId={}, requestType={}, req={}", + PipeDataLossDebugUtil.PREFIX, + receiverId.get(), + requestType, + PipeDataLossDebugUtil.formatReq(req)); + } if (requestType != PipeRequestType.TRANSFER_SLICE) { clearSliceReqHandler(); } @@ -487,51 +496,105 @@ public synchronized TPipeTransferResp receive(final TPipeTransferReq req) { private TPipeTransferResp handleTransferTabletInsertNode( final PipeTransferTabletInsertNodeReq req) { final InsertBaseStatement statement = req.constructStatement(); - return new TPipeTransferResp( + final TSStatus status = statement.isEmpty() ? RpcUtils.SUCCESS_STATUS - : executeStatementAndClassifyExceptions(statement)); + : executeStatementAndClassifyExceptions(statement); + logReceiverStatement("insert-node", statement, status); + return new TPipeTransferResp(status); } private TPipeTransferResp handleTransferTabletBinary(final PipeTransferTabletBinaryReq req) { final InsertBaseStatement statement = req.constructStatement(); - return new TPipeTransferResp( + final TSStatus status = statement.isEmpty() ? RpcUtils.SUCCESS_STATUS - : executeStatementAndClassifyExceptions(statement)); + : executeStatementAndClassifyExceptions(statement); + logReceiverStatement("binary", statement, status); + return new TPipeTransferResp(status); } private TPipeTransferResp handleTransferTabletRaw(final PipeTransferTabletRawReq req) { final InsertTabletStatement statement = req.constructStatement(); - return new TPipeTransferResp( + final TSStatus status = statement.isEmpty() ? RpcUtils.SUCCESS_STATUS - : executeStatementAndClassifyExceptions(statement)); + : executeStatementAndClassifyExceptions(statement); + logReceiverStatement("raw", statement, status); + return new TPipeTransferResp(status); + } + + private void logReceiverStatement( + final String requestType, final Statement statement, final TSStatus status) { + if (!LOGGER.isDebugEnabled()) { + return; + } + + LOGGER.debug( + "{} receiver executed tablet {}, statement={}, status={}, mapping={}", + PipeDataLossDebugUtil.PREFIX, + requestType, + PipeDataLossDebugUtil.formatStatement(statement), + PipeDataLossDebugUtil.formatStatus(status), + PipeDataLossDebugUtil.formatStatementStatusMapping(statement, status)); } private TPipeTransferResp handleTransferTabletBatch(final PipeTransferTabletBatchReq req) { final Pair statementPair = req.constructStatements(); - return new TPipeTransferResp( - PipeReceiverStatusHandler.getPriorStatus( - Stream.of( - statementPair.getLeft().isEmpty() - ? RpcUtils.SUCCESS_STATUS - : executeBatchStatementAndAddRedirectInfo(statementPair.getLeft()), - statementPair.getRight().isEmpty() - ? RpcUtils.SUCCESS_STATUS - : executeBatchStatementAndAddRedirectInfo(statementPair.getRight())) - .collect(Collectors.toList()))); + final List statusList = + Stream.of( + statementPair.getLeft().isEmpty() + ? RpcUtils.SUCCESS_STATUS + : executeBatchStatementAndAddRedirectInfo(statementPair.getLeft()), + statementPair.getRight().isEmpty() + ? RpcUtils.SUCCESS_STATUS + : executeBatchStatementAndAddRedirectInfo(statementPair.getRight())) + .collect(Collectors.toList()); + final TSStatus priorStatus = PipeReceiverStatusHandler.getPriorStatus(statusList); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver executed tablet batch, statements={}, statuses={}, priorStatus={}, rowMapping={}, tabletMapping={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatStatements( + Arrays.asList(statementPair.getLeft(), statementPair.getRight())), + PipeDataLossDebugUtil.formatStatuses(statusList), + PipeDataLossDebugUtil.formatStatus(priorStatus), + PipeDataLossDebugUtil.formatStatementStatusMapping( + statementPair.getLeft(), statusList.get(0)), + PipeDataLossDebugUtil.formatStatementStatusMapping( + statementPair.getRight(), statusList.get(1))); + } + return new TPipeTransferResp(priorStatus); } private TPipeTransferResp handleTransferTabletBatchV2(final PipeTransferTabletBatchReqV2 req) { final List statementSet = req.constructStatements(); - return new TPipeTransferResp( - PipeReceiverStatusHandler.getPriorStatus( - (statementSet.isEmpty() - ? Stream.of(RpcUtils.SUCCESS_STATUS) - : statementSet.stream().map(this::executeBatchStatementAndAddRedirectInfo)) - .collect(Collectors.toList()))); + final List statusList = + (statementSet.isEmpty() + ? Stream.of(RpcUtils.SUCCESS_STATUS) + : statementSet.stream().map(this::executeBatchStatementAndAddRedirectInfo)) + .collect(Collectors.toList()); + final TSStatus priorStatus = PipeReceiverStatusHandler.getPriorStatus(statusList); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver executed tablet batch V2, statements={}, statuses={}, priorStatus={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatStatements(statementSet), + PipeDataLossDebugUtil.formatStatuses(statusList), + PipeDataLossDebugUtil.formatStatus(priorStatus)); + for (int i = 0; i < statementSet.size() && i < statusList.size(); ++i) { + LOGGER.debug( + "{} receiver tablet batch V2 statement result, index={}, statement={}, status={}, mapping={}", + PipeDataLossDebugUtil.PREFIX, + i, + PipeDataLossDebugUtil.formatStatement(statementSet.get(i)), + PipeDataLossDebugUtil.formatStatus(statusList.get(i)), + PipeDataLossDebugUtil.formatStatementStatusMapping( + statementSet.get(i), statusList.get(i))); + } + } + return new TPipeTransferResp(priorStatus); } @Override @@ -935,7 +998,23 @@ protected TSStatus getReceiverTemporaryUnavailableStatus( * message field. */ private TSStatus executeBatchStatementAndAddRedirectInfo(final InsertBaseStatement statement) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver executing batch statement, receiverId={}, statement={}", + PipeDataLossDebugUtil.PREFIX, + receiverId.get(), + PipeDataLossDebugUtil.formatStatement(statement)); + } final TSStatus result = executeStatementAndClassifyExceptions(statement, 5); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver executed batch statement before redirection decoration, receiverId={}, status={}, mapping={}, statement={}", + PipeDataLossDebugUtil.PREFIX, + receiverId.get(), + PipeDataLossDebugUtil.formatStatus(result), + PipeDataLossDebugUtil.formatStatementStatusMapping(statement, result), + PipeDataLossDebugUtil.formatStatement(statement)); + } if (result.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode() && result.getSubStatusSize() > 0) { @@ -967,6 +1046,15 @@ private TSStatus executeBatchStatementAndAddRedirectInfo(final InsertBaseStateme } } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver finalized batch statement status, receiverId={}, status={}, mapping={}, statement={}", + PipeDataLossDebugUtil.PREFIX, + receiverId.get(), + PipeDataLossDebugUtil.formatStatus(result), + PipeDataLossDebugUtil.formatStatementStatusMapping(statement, result), + PipeDataLossDebugUtil.formatStatement(statement)); + } return result; } @@ -980,8 +1068,26 @@ private TSStatus executeStatementAndClassifyExceptions( final double pipeReceiverActualToEstimatedMemoryRatio = PIPE_CONFIG.getPipeReceiverActualToEstimatedMemoryRatio(); try { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver executeStatement start, receiverId={}, tryCount={}, statement={}", + PipeDataLossDebugUtil.PREFIX, + receiverId.get(), + tryCount, + PipeDataLossDebugUtil.formatStatement(statement)); + } if (statement instanceof InsertBaseStatement) { estimatedMemory = ((InsertBaseStatement) statement).ramBytesUsed(); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver estimated statement memory, receiverId={}, estimatedMemory={}, actualToEstimatedRatio={}, requestedMemory={}, statement={}", + PipeDataLossDebugUtil.PREFIX, + receiverId.get(), + estimatedMemory, + pipeReceiverActualToEstimatedMemoryRatio, + estimatedMemory * pipeReceiverActualToEstimatedMemoryRatio, + PipeDataLossDebugUtil.formatStatement(statement)); + } for (int i = 0; i < tryCount; ++i) { try { allocatedMemoryBlock = @@ -1016,6 +1122,15 @@ private TSStatus executeStatementAndClassifyExceptions( final TSStatus result = executeStatementWithPermissionCheckAndRetryOnDataTypeMismatch(statement); final int code = result.getCode(); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver executeStatement raw result, receiverId={}, status={}, mapping={}, statement={}", + PipeDataLossDebugUtil.PREFIX, + receiverId.get(), + PipeDataLossDebugUtil.formatStatus(result), + PipeDataLossDebugUtil.formatStatementStatusMapping(statement, result), + PipeDataLossDebugUtil.formatStatement(statement)); + } if (code == TSStatusCode.SUCCESS_STATUS.getStatusCode() || code == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { return result; @@ -1028,11 +1143,38 @@ private TSStatus executeStatementAndClassifyExceptions( statement.getPipeLoggingString(), result); } - return STATEMENT_STATUS_VISITOR.process(statement, result); + final TSStatus classifiedStatus = STATEMENT_STATUS_VISITOR.process(statement, result); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver executeStatement classified failure status, receiverId={}, rawStatus={}, classifiedStatus={}, statement={}", + PipeDataLossDebugUtil.PREFIX, + receiverId.get(), + PipeDataLossDebugUtil.formatStatus(result), + PipeDataLossDebugUtil.formatStatus(classifiedStatus), + PipeDataLossDebugUtil.formatStatement(statement)); + } + return classifiedStatus; } } catch (final Exception e) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver executeStatement exception, receiverId={}, exception={}, statement={}", + PipeDataLossDebugUtil.PREFIX, + receiverId.get(), + PipeDataLossDebugUtil.formatException(e), + PipeDataLossDebugUtil.formatStatement(statement)); + } logStatementExceptionIfNecessary(statement, e); - return STATEMENT_EXCEPTION_VISITOR.process(statement, e); + final TSStatus classifiedStatus = STATEMENT_EXCEPTION_VISITOR.process(statement, e); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} receiver executeStatement classified exception status, receiverId={}, status={}, statement={}", + PipeDataLossDebugUtil.PREFIX, + receiverId.get(), + PipeDataLossDebugUtil.formatStatus(classifiedStatus), + PipeDataLossDebugUtil.formatStatement(statement)); + } + return classifiedStatus; } finally { if (Objects.nonNull(allocatedMemoryBlock)) { allocatedMemoryBlock.close(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java index b850de896a1a3..f700ab32f5264 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.db.i18n.DataNodePipeMessages; +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.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; @@ -81,6 +82,13 @@ protected PipeTabletEventBatch( public synchronized boolean onEvent(final TabletInsertionEvent event) throws WALPipeException, IOException { if (isClosed || !(event instanceof EnrichedEvent)) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} tablet batch ignored event, isClosed={}, event={}", + PipeDataLossDebugUtil.PREFIX, + isClosed, + String.valueOf(event)); + } return false; } @@ -92,10 +100,28 @@ public synchronized boolean onEvent(final TabletInsertionEvent event) .increaseReferenceCount(PipeTransferBatchReqBuilder.class.getName())) { try { - if (constructBatch(event)) { + final boolean constructed = constructBatch(event); + if (constructed) { events.add((EnrichedEvent) event); } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} tablet batch construct result, constructed={}, event={}, batchEvents={}, totalBufferSize={}", + PipeDataLossDebugUtil.PREFIX, + constructed, + PipeDataLossDebugUtil.formatEvent((EnrichedEvent) event), + PipeDataLossDebugUtil.formatEvents(events), + totalBufferSize); + } } catch (final Exception e) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} tablet batch construct failed, event={}, batchEvents={}, exception={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent((EnrichedEvent) event), + PipeDataLossDebugUtil.formatEvents(events), + PipeDataLossDebugUtil.formatException(e)); + } if (events.isEmpty()) { clearBatchData(); resetMemoryUsage(); @@ -159,6 +185,13 @@ public boolean shouldEmit() { } public synchronized void onSuccess() { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} tablet batch onSuccess clearing batch data, events={}, totalBufferSize={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvents(events), + totalBufferSize); + } events.clear(); resetMemoryUsage(); @@ -171,6 +204,13 @@ public synchronized void close() { } isClosed = true; + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} tablet batch closing, events={}, totalBufferSize={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvents(events), + totalBufferSize); + } clearEventsReferenceCount(PipeTabletEventBatch.class.getName()); events.clear(); clearBatchData(); @@ -188,6 +228,13 @@ public synchronized void discardEventsOfPipe( } public synchronized void discardEventsOfPipe(final CommitterKey committerKey) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} tablet batch discarding events of pipe, committerKey={}, beforeEvents={}", + PipeDataLossDebugUtil.PREFIX, + committerKey, + PipeDataLossDebugUtil.formatEvents(events)); + } final boolean hasDiscardedEvents = events.removeIf( event -> { @@ -197,6 +244,14 @@ public synchronized void discardEventsOfPipe(final CommitterKey committerKey) { } return false; }); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} tablet batch discarded events of pipe, committerKey={}, hasDiscardedEvents={}, afterEvents={}", + PipeDataLossDebugUtil.PREFIX, + committerKey, + hasDiscardedEvents, + PipeDataLossDebugUtil.formatEvents(events)); + } if (hasDiscardedEvents && events.isEmpty()) { clearBatchData(); resetMemoryUsage(); @@ -221,10 +276,25 @@ private static boolean isEventFromPipe( public synchronized void decreaseEventsReferenceCount( final String holderMessage, final boolean shouldReport) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} tablet batch decreasing event references, holder={}, shouldReport={}, events={}", + PipeDataLossDebugUtil.PREFIX, + holderMessage, + shouldReport, + PipeDataLossDebugUtil.formatEvents(events)); + } events.forEach(event -> event.decreaseReferenceCount(holderMessage, shouldReport)); } private void clearEventsReferenceCount(final String holderMessage) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} tablet batch clearing event references, holder={}, events={}", + PipeDataLossDebugUtil.PREFIX, + holderMessage, + PipeDataLossDebugUtil.formatEvents(events)); + } events.forEach(event -> event.clearReferenceCount(holderMessage)); } @@ -236,6 +306,14 @@ public boolean isEmpty() { return events.isEmpty(); } + public int getEventCount() { + return events.size(); + } + + public long getTotalBufferSize() { + return totalBufferSize; + } + @FunctionalInterface public interface TriLongConsumer { void accept(long l1, long l2, long l3); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java index ce70cf6f6e33e..6ee29b159b83a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java @@ -25,6 +25,7 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.util.PipeDataLossDebugUtil; import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeCacheLeaderClientManager; import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException; import org.apache.iotdb.metrics.impl.DoNothingHistogram; @@ -146,7 +147,8 @@ public synchronized void onEvent(final TabletInsertionEvent event) } if (!useLeaderCache) { - defaultBatch.onEvent(event); + final boolean shouldEmit = defaultBatch.onEvent(event); + logOnEvent(event, "default", null, defaultBatch, shouldEmit); return; } @@ -158,23 +160,52 @@ public synchronized void onEvent(final TabletInsertionEvent event) } if (Objects.isNull(deviceId)) { - defaultBatch.onEvent(event); + final boolean shouldEmit = defaultBatch.onEvent(event); + logOnEvent(event, "default-no-device", null, defaultBatch, shouldEmit); return; } final TEndPoint endPoint = IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId); if (Objects.isNull(endPoint)) { - defaultBatch.onEvent(event); + final boolean shouldEmit = defaultBatch.onEvent(event); + logOnEvent(event, "default-no-leader", null, defaultBatch, shouldEmit); return; } - endPointToBatch - .computeIfAbsent( + final PipeTabletEventPlainBatch batch = + endPointToBatch.computeIfAbsent( endPoint, k -> new PipeTabletEventPlainBatch( - requestMaxDelayInMs, requestMaxBatchSizeInBytes, this::recordTabletMetric)) - .onEvent(event); + requestMaxDelayInMs, requestMaxBatchSizeInBytes, this::recordTabletMetric)); + final boolean shouldEmit = batch.onEvent(event); + logOnEvent(event, "leader-cache", endPoint, batch, shouldEmit); + } + + private void logOnEvent( + final TabletInsertionEvent event, + final String route, + final TEndPoint endPoint, + final PipeTabletEventBatch batch, + final boolean shouldEmit) { + if (!LOGGER.isDebugEnabled() || !(event instanceof EnrichedEvent)) { + return; + } + + final EnrichedEvent enrichedEvent = (EnrichedEvent) event; + LOGGER.debug( + "{} sink batch accepted tablet, {}, regionId={}, route={}, endPoint={}, shouldEmit={}, batchEventCount={}, batchBufferSize={}, event={}, batchEvents={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatPipe( + enrichedEvent.getPipeName(), enrichedEvent.getCreationTime()), + enrichedEvent.getRegionId(), + route, + endPoint, + shouldEmit, + batch.getEventCount(), + batch.getTotalBufferSize(), + PipeDataLossDebugUtil.formatTabletInsertionEvent(event), + PipeDataLossDebugUtil.formatEvents(batch.deepCopyEvents())); } /** Get all batches that have at least 1 event. */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java index 250ebeabb39df..8cceb1786a8ab 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java @@ -37,6 +37,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.metric.sink.PipeDataRegionSinkMetrics; import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter; import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeAsyncClientManager; @@ -249,6 +250,7 @@ private void transferInBatchWithoutCheck( } final PipeTabletEventBatch batch = endPointAndBatch.getRight(); + logTransferBatch(endPointAndBatch.getLeft(), batch); if (batch instanceof PipeTabletEventPlainBatch) { transfer( @@ -293,9 +295,33 @@ private void transferInBatchWithoutCheck( batch.getClass()); } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink marking emitted batch as locally successful and clearing builder batch, endPoint={}, batchType={}, events={}", + PipeDataLossDebugUtil.PREFIX, + endPointAndBatch.getLeft(), + batch.getClass().getSimpleName(), + PipeDataLossDebugUtil.formatEvents(batch.deepCopyEvents())); + } endPointAndBatch.getRight().onSuccess(); } + private void logTransferBatch(final TEndPoint endPoint, final PipeTabletEventBatch batch) { + if (!LOGGER.isDebugEnabled()) { + return; + } + + final List events = batch.deepCopyEvents(); + LOGGER.debug( + "{} async sink emitting tablet batch, endPoint={}, batchType={}, eventCount={}, batchBufferSize={}, events={}", + PipeDataLossDebugUtil.PREFIX, + endPoint, + batch.getClass().getSimpleName(), + batch.getEventCount(), + batch.getTotalBufferSize(), + PipeDataLossDebugUtil.formatEvents(events)); + } + private boolean transferInEventWithoutCheck(final TabletInsertionEvent tabletInsertionEvent) throws Exception { if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) { @@ -356,8 +382,24 @@ private void transfer( try { client = clientManager.borrowClient(endPoint); markHandshakeSucceeded(); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink borrowed client for tablet batch, requestedEndPoint={}, actualEndPoint={}, pendingHandlers={}", + PipeDataLossDebugUtil.PREFIX, + endPoint, + client.getEndPoint(), + getPendingHandlersSize()); + } pipeTransferTabletBatchEventHandler.transfer(client); } catch (final Exception ex) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink failed before/during tablet batch transfer, requestedEndPoint={}, actualEndPoint={}, exception={}", + PipeDataLossDebugUtil.PREFIX, + endPoint, + client == null ? null : client.getEndPoint(), + PipeDataLossDebugUtil.formatException(ex)); + } markSchedulingDelayIfHandshakeFailed(client); logOnClientException(client, ex); pipeTransferTabletBatchEventHandler.onError(ex); @@ -606,6 +648,17 @@ public long consumeSchedulingDelayMs() { * @see PipeConnector#transfer(TsFileInsertionEvent) for more details. */ private void transferQueuedEventsIfNecessary(final boolean forced) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink checking retry queues, forced={}, {}", + PipeDataLossDebugUtil.PREFIX, + forced, + PipeDataLossDebugUtil.formatQueueState( + retryEventQueue.size(), + retryTsFileQueue.size(), + retryEventQueueEventCounter.getTabletInsertionEventCount(), + retryEventQueueEventCounter.getTsFileInsertionEventCount())); + } if ((retryEventQueue.isEmpty() && retryTsFileQueue.isEmpty()) || (!forced && retryEventQueueEventCounter.getTabletInsertionEventCount() @@ -632,6 +685,17 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { final Event polledEvent; if (!retryEventQueue.isEmpty()) { peekedEvent = retryEventQueue.peek(); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink retrying tablet event from retry queue, event={}, {}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent(peekedEvent), + PipeDataLossDebugUtil.formatQueueState( + retryEventQueue.size(), + retryTsFileQueue.size(), + retryEventQueueEventCounter.getTabletInsertionEventCount(), + retryEventQueueEventCounter.getTsFileInsertionEventCount())); + } if (peekedEvent instanceof PipeInsertNodeTabletInsertionEvent) { retryTransfer((PipeInsertNodeTabletInsertionEvent) peekedEvent); @@ -651,6 +715,17 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { return; } peekedEvent = retryTsFileQueue.peek(); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink retrying tsfile event from retry queue, event={}, {}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent(peekedEvent), + PipeDataLossDebugUtil.formatQueueState( + retryEventQueue.size(), + retryTsFileQueue.size(), + retryEventQueueEventCounter.getTabletInsertionEventCount(), + retryEventQueueEventCounter.getTsFileInsertionEventCount())); + } retryTransfer((PipeTsFileInsertionEvent) peekedEvent); polledEvent = retryTsFileQueue.poll(); } @@ -662,6 +737,15 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { } if (polledEvent != null && LOGGER.isDebugEnabled()) { LOGGER.debug(DataNodePipeMessages.POLLED_EVENT_FROM_RETRY_QUEUE, polledEvent); + LOGGER.debug( + "{} async sink polled event from retry queue after retry attempt, polledEvent={}, {}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent(polledEvent), + PipeDataLossDebugUtil.formatQueueState( + retryEventQueue.size(), + retryTsFileQueue.size(), + retryEventQueueEventCounter.getTabletInsertionEventCount(), + retryEventQueueEventCounter.getTsFileInsertionEventCount())); } } @@ -695,11 +779,24 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { } private void retryTransfer(final TabletInsertionEvent tabletInsertionEvent) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink retryTransfer tablet start, batchMode={}, event={}", + PipeDataLossDebugUtil.PREFIX, + isTabletBatchModeEnabled, + PipeDataLossDebugUtil.formatEvent((Event) tabletInsertionEvent)); + } if (isTabletBatchModeEnabled) { try { tabletBatchBuilder.onEvent(tabletInsertionEvent); transferBatchedEventsIfNecessary(); if (tabletInsertionEvent instanceof EnrichedEvent) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink retryTransfer tablet batched successfully, decrease retry holder reference, event={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent((Event) tabletInsertionEvent)); + } ((EnrichedEvent) tabletInsertionEvent) .decreaseReferenceCount(IoTDBDataRegionAsyncSink.class.getName(), false); } @@ -729,6 +826,12 @@ private void retryTransfer(final TabletInsertionEvent tabletInsertionEvent) { } private void retryTransfer(final PipeTsFileInsertionEvent tsFileInsertionEvent) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink retryTransfer tsfile start, event={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent(tsFileInsertionEvent)); + } try { if (transferWithoutCheck(tsFileInsertionEvent)) { tsFileInsertionEvent.decreaseReferenceCount( @@ -750,18 +853,49 @@ private void retryTransfer(final PipeTsFileInsertionEvent tsFileInsertionEvent) public void addFailureEventToRetryQueue(final Event event, final Exception e) { isConnectionException = e instanceof PipeConnectionException || ThriftClient.isConnectionBroken(e); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink addFailureEventToRetryQueue invoked, event={}, exception={}, isConnectionException={}, {}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent(event), + PipeDataLossDebugUtil.formatException(e), + isConnectionException, + PipeDataLossDebugUtil.formatQueueState( + retryEventQueue.size(), + retryTsFileQueue.size(), + retryEventQueueEventCounter.getTabletInsertionEventCount(), + retryEventQueueEventCounter.getTsFileInsertionEventCount())); + } if (event instanceof EnrichedEvent) { final EnrichedEvent enrichedEvent = (EnrichedEvent) event; if (enrichedEvent.isReleased()) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink skip adding released event to retry queue, event={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent(event)); + } return; } if (isDroppedPipe(enrichedEvent)) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink skip adding dropped-pipe event to retry queue and clear reference, event={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent(event)); + } enrichedEvent.clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); return; } } if (isClosed.get()) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink skip adding event to retry queue because sink is closed, event={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent(event)); + } if (event instanceof EnrichedEvent) { ((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); } @@ -778,9 +912,24 @@ public void addFailureEventToRetryQueue(final Event event, final Exception e) { if (LOGGER.isDebugEnabled()) { LOGGER.debug(DataNodePipeMessages.ADDED_EVENT_TO_RETRY_QUEUE, event); + LOGGER.debug( + "{} async sink added event to retry queue, event={}, {}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent(event), + PipeDataLossDebugUtil.formatQueueState( + retryEventQueue.size(), + retryTsFileQueue.size(), + retryEventQueueEventCounter.getTabletInsertionEventCount(), + retryEventQueueEventCounter.getTsFileInsertionEventCount())); } if (isClosed.get()) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink became closed after adding event to retry queue, clear reference, event={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvent(event)); + } if (event instanceof EnrichedEvent) { ((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); } @@ -794,6 +943,13 @@ public void addFailureEventToRetryQueue(final Event event, final Exception e) { */ public void addFailureEventsToRetryQueue( final Iterable events, final Exception e) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink addFailureEventsToRetryQueue invoked, events={}, exception={}", + PipeDataLossDebugUtil.PREFIX, + PipeDataLossDebugUtil.formatEvents(events), + PipeDataLossDebugUtil.formatException(e)); + } events.forEach(event -> addFailureEventToRetryQueue(event, e)); } @@ -819,6 +975,13 @@ public void waitIfReceiverTemporarilyUnavailable(final TEndPoint endPoint) { } try { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink throttling because receiver is temporarily unavailable, endPoint={}, waitTimeInMs={}", + PipeDataLossDebugUtil.PREFIX, + endPoint, + waitTimeInMs); + } Thread.sleep(waitTimeInMs); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); @@ -828,6 +991,13 @@ public void waitIfReceiverTemporarilyUnavailable(final TEndPoint endPoint) { } public void recordReceiverStatus(final TEndPoint endPoint, final TSStatus status) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink recording receiver status, endPoint={}, status={}", + PipeDataLossDebugUtil.PREFIX, + endPoint, + PipeDataLossDebugUtil.formatStatus(status)); + } final String endPointKey = format(endPoint); if (Objects.isNull(endPointKey) || Objects.isNull(status)) { return; @@ -991,6 +1161,13 @@ public boolean isClosed() { public void trackHandler(final PipeTransferTrackableHandler handler) { pendingHandlers.put(handler, handler); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink tracked handler, handler={}, pendingHandlers={}", + PipeDataLossDebugUtil.PREFIX, + handler.getClass().getSimpleName() + "@" + System.identityHashCode(handler), + pendingHandlers.size()); + } } public void eliminateHandler( @@ -1000,6 +1177,14 @@ public void eliminateHandler( } handler.close(); pendingHandlers.remove(handler); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async sink eliminated handler, handler={}, closeClient={}, pendingHandlers={}", + PipeDataLossDebugUtil.PREFIX, + handler.getClass().getSimpleName() + "@" + System.identityHashCode(handler), + closeClient, + pendingHandlers.size()); + } } public boolean hasPendingHandlers() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java index c58a828625b73..0b31cab6cc707 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.db.i18n.DataNodePipeMessages; +import org.apache.iotdb.db.pipe.event.common.util.PipeDataLossDebugUtil; import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventPlainBatch; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; import org.apache.iotdb.db.pipe.sink.util.cacher.LeaderCacheUtils; @@ -41,6 +42,7 @@ import java.io.IOException; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; public class PipeTransferTabletBatchEventHandler extends PipeTransferTrackableHandler { @@ -48,10 +50,15 @@ public class PipeTransferTabletBatchEventHandler extends PipeTransferTrackableHa private static final Logger LOGGER = LoggerFactory.getLogger(PipeTransferTabletBatchEventHandler.class); + private static final AtomicLong DEBUG_BATCH_ID_GENERATOR = new AtomicLong(0); + + private final long debugBatchId; private final List events; private final Map, Long> pipeName2BytesAccumulated; private final TPipeTransferReq req; + private final String uncompressedReqDebugInfo; + private final String compressedReqDebugInfo; private final double reqCompressionRatio; public PipeTransferTabletBatchEventHandler( @@ -59,16 +66,43 @@ public PipeTransferTabletBatchEventHandler( throws IOException { super(connector); + debugBatchId = DEBUG_BATCH_ID_GENERATOR.incrementAndGet(); + // Deep copy to keep events' reference events = batch.deepCopyEvents(); pipeName2BytesAccumulated = batch.deepCopyPipeName2BytesAccumulated(); final TPipeTransferReq uncompressedReq = batch.toTPipeTransferReq(); + uncompressedReqDebugInfo = PipeDataLossDebugUtil.formatReq(uncompressedReq); req = connector.compressIfNeeded(uncompressedReq); + compressedReqDebugInfo = PipeDataLossDebugUtil.formatReq(req); reqCompressionRatio = (double) req.getBody().length / uncompressedReq.getBody().length; + + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} sender built tablet batch request, debugBatchId={}, uncompressedReq={}, transferredReq={}, compressionRatio={}, events={}, bytesByPipe={}", + PipeDataLossDebugUtil.PREFIX, + debugBatchId, + uncompressedReqDebugInfo, + compressedReqDebugInfo, + reqCompressionRatio, + PipeDataLossDebugUtil.formatEvents(events), + pipeName2BytesAccumulated); + } } public void transfer(final AsyncPipeDataTransferServiceClient client) throws TException { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} sender transferring tablet batch, debugBatchId={}, endpoint={}, transferredReq={}, events={}, bytesByPipe={}", + PipeDataLossDebugUtil.PREFIX, + debugBatchId, + client.getEndPoint(), + compressedReqDebugInfo, + PipeDataLossDebugUtil.formatEvents(events), + pipeName2BytesAccumulated); + } + for (final Map.Entry, Long> entry : pipeName2BytesAccumulated.entrySet()) { sink.rateLimitIfNeeded( entry.getKey().getLeft(), @@ -84,12 +118,30 @@ public void transfer(final AsyncPipeDataTransferServiceClient client) throws TEx protected boolean onCompleteInternal(final TPipeTransferResp response) { // Just in case if (response == null) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} sender tablet batch callback got null response, debugBatchId={}, endpoint={}, events={}", + PipeDataLossDebugUtil.PREFIX, + debugBatchId, + client == null ? null : client.getEndPoint(), + PipeDataLossDebugUtil.formatEvents(events)); + } onError(new PipeException(DataNodePipeMessages.TPIPETRANSFERRESP_IS_NULL)); return false; } try { final TSStatus status = response.getStatus(); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} sender tablet batch callback complete, debugBatchId={}, endpoint={}, status={}, transferredReq={}, events={}", + PipeDataLossDebugUtil.PREFIX, + debugBatchId, + client == null ? null : client.getEndPoint(), + PipeDataLossDebugUtil.formatStatus(status), + compressedReqDebugInfo, + PipeDataLossDebugUtil.formatEvents(events)); + } // Only handle the failed statuses to avoid string format performance overhead if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { @@ -100,11 +152,27 @@ protected boolean onCompleteInternal(final TPipeTransferResp response) { sink.updateLeaderCache(redirectPair.getLeft(), redirectPair.getRight()); } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} sender tablet batch decreasing event references after callback, debugBatchId={}, shouldReport=true, events={}", + PipeDataLossDebugUtil.PREFIX, + debugBatchId, + PipeDataLossDebugUtil.formatEvents(events)); + } events.forEach( event -> event.decreaseReferenceCount( PipeTransferTabletBatchEventHandler.class.getName(), true)); } catch (final Exception e) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} sender tablet batch callback handling failed, debugBatchId={}, endpoint={}, exception={}, events={}", + PipeDataLossDebugUtil.PREFIX, + debugBatchId, + client == null ? null : client.getEndPoint(), + PipeDataLossDebugUtil.formatException(e), + PipeDataLossDebugUtil.formatEvents(events)); + } onError(e); return false; } @@ -115,6 +183,15 @@ protected boolean onCompleteInternal(final TPipeTransferResp response) { @Override protected void onErrorInternal(final Exception exception) { try { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} sender tablet batch transfer failed, debugBatchId={}, endpoint={}, exception={}, events={}", + PipeDataLossDebugUtil.PREFIX, + debugBatchId, + client == null ? null : client.getEndPoint(), + PipeDataLossDebugUtil.formatException(exception), + PipeDataLossDebugUtil.formatEvents(events)); + } PipeLogger.log( LOGGER::warn, exception, @@ -122,6 +199,14 @@ protected void onErrorInternal(final Exception exception) { events.size(), events.stream().map(EnrichedEvent::getPipeName).collect(Collectors.toSet())); } finally { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} sender tablet batch adding events to retry queue, debugBatchId={}, exception={}, events={}", + PipeDataLossDebugUtil.PREFIX, + debugBatchId, + PipeDataLossDebugUtil.formatException(exception), + PipeDataLossDebugUtil.formatEvents(events)); + } sink.addFailureEventsToRetryQueue(events, exception); } } @@ -135,7 +220,25 @@ protected void doTransfer( @Override public void clearEventsReferenceCount() { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} sender tablet batch clearing event references, debugBatchId={}, events={}", + PipeDataLossDebugUtil.PREFIX, + debugBatchId, + PipeDataLossDebugUtil.formatEvents(events)); + } events.forEach( event -> event.clearReferenceCount(PipeTransferTabletBatchEventHandler.class.getName())); } + + @Override + protected String getDebugHandlerInfo() { + return "debugBatchId=" + + debugBatchId + + ", transferredReq={" + + compressedReqDebugInfo + + "}, events={" + + PipeDataLossDebugUtil.formatEvents(events) + + "}"; + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java index 7ec25753ea749..076a7aa69752a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferSliceReqBuilder; import org.apache.iotdb.db.i18n.DataNodePipeMessages; +import org.apache.iotdb.db.pipe.event.common.util.PipeDataLossDebugUtil; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; import org.apache.iotdb.pipe.api.exception.PipeConnectionException; import org.apache.iotdb.pipe.api.exception.PipeException; @@ -51,11 +52,26 @@ public PipeTransferTrackableHandler(final IoTDBDataRegionAsyncSink sink) { @Override public void onComplete(final TPipeTransferResp response) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler onComplete, handler={}, endpoint={}, responseStatus={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client == null ? null : client.getEndPoint(), + response == null ? null : PipeDataLossDebugUtil.formatStatus(response.getStatus())); + } + if (Objects.nonNull(client) && Objects.nonNull(response)) { sink.recordReceiverStatus(client.getEndPoint(), response.getStatus()); } if (sink.isClosed()) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler onComplete found closed sink, clearing references, handler={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo()); + } clearEventsReferenceCount(); sink.eliminateHandler(this, true); return; @@ -72,12 +88,27 @@ public void onComplete(final TPipeTransferResp response) { @Override public void onError(final Exception exception) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler onError, handler={}, endpoint={}, exception={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client == null ? null : client.getEndPoint(), + PipeDataLossDebugUtil.formatException(exception)); + } + if (client != null) { ThriftClient.resolveException(exception, client); client.setPrintLogWhenEncounterException(false); } if (sink.isClosed()) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler onComplete found closed sink, clearing references, handler={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo()); + } clearEventsReferenceCount(); sink.eliminateHandler(this, true); return; @@ -104,13 +135,37 @@ protected boolean tryTransfer( } // track handler before checking if connector is closed sink.trackHandler(this); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler tracked before transfer, handler={}, endpoint={}, req={}, pendingHandlers={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client.getEndPoint(), + PipeDataLossDebugUtil.formatReq(req), + sink.getPendingHandlersSize()); + } if (returnFalseIfSinkIsClosed(client)) { return false; } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler waiting for receiver availability if needed, handler={}, endpoint={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client.getEndPoint()); + } sink.waitIfReceiverTemporarilyUnavailable(client.getEndPoint()); if (returnFalseIfSinkIsClosed(client)) { return false; } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler doing transfer, handler={}, endpoint={}, req={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client.getEndPoint(), + PipeDataLossDebugUtil.formatReq(req)); + } doTransfer(client, req); return true; } @@ -120,6 +175,13 @@ private boolean returnFalseIfSinkIsClosed(final AsyncPipeDataTransferServiceClie return false; } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler found closed sink before transfer, clearing references, handler={}, endpoint={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client.getEndPoint()); + } clearEventsReferenceCount(); sink.eliminateHandler(this, true); client.setShouldReturnSelf(true); @@ -155,6 +217,15 @@ protected final void transferWithOptionalRequestSlicing( throws TException { final int bodySizeLimit = PipeTransferSliceReqBuilder.getBodySizeLimit(); if (!PipeTransferSliceReqBuilder.shouldSlice(req, bodySizeLimit)) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler transferring whole request, handler={}, endpoint={}, req={}, bodySizeLimit={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client.getEndPoint(), + PipeDataLossDebugUtil.formatReq(req), + bodySizeLimit); + } client.pipeTransfer(req, this); return; } @@ -168,6 +239,16 @@ protected final void transferWithOptionalRequestSlicing( bodySizeLimit); final int sliceCount = PipeTransferSliceReqBuilder.getSliceCount(req, bodySizeLimit); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler transferring sliced request, handler={}, endpoint={}, req={}, sliceCount={}, bodySizeLimit={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client.getEndPoint(), + PipeDataLossDebugUtil.formatReq(req), + sliceCount, + bodySizeLimit); + } final boolean shouldReturnSelf = client.shouldReturnSelf(); try { transferSlicedRequest( @@ -183,6 +264,13 @@ protected final void transferWithOptionalRequestSlicing( } } + protected String getDebugHandlerInfo() { + return "handlerClass=" + + getClass().getSimpleName() + + ", identity=" + + System.identityHashCode(this); + } + public abstract void clearEventsReferenceCount(); private void transferSlicedRequest( @@ -207,6 +295,16 @@ public void onComplete(final TPipeTransferResp response) { } if (Objects.nonNull(response)) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler slice callback, handler={}, endpoint={}, sliceIndex={}, sliceCount={}, status={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client.getEndPoint(), + sliceIndex, + sliceCount, + PipeDataLossDebugUtil.formatStatus(response.getStatus())); + } sink.recordReceiverStatus(client.getEndPoint(), response.getStatus()); } @@ -257,6 +355,16 @@ public void onError(final Exception exception) { PipeTransferTrackableHandler.this.onError(exception); return; } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler slice onError, handler={}, endpoint={}, sliceIndex={}, sliceCount={}, exception={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client.getEndPoint(), + sliceIndex, + sliceCount, + PipeDataLossDebugUtil.formatException(exception)); + } fallbackToWholeRequest(client, originalReq, shouldReturnSelf, exception); } }); @@ -267,6 +375,15 @@ private void fallbackToWholeRequest( final TPipeTransferReq originalReq, final boolean shouldReturnSelf, final Exception exception) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} async handler falling back to whole request, handler={}, endpoint={}, req={}, exception={}", + PipeDataLossDebugUtil.PREFIX, + getDebugHandlerInfo(), + client.getEndPoint(), + PipeDataLossDebugUtil.formatReq(originalReq), + PipeDataLossDebugUtil.formatException(exception)); + } PipeLogger.log( LOGGER::warn, exception, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java index d13b9664f8ab2..8a8b514993e55 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java @@ -36,6 +36,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.metric.overview.PipeResourceMetrics; import org.apache.iotdb.db.pipe.metric.sink.PipeDataRegionSinkMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; @@ -285,6 +286,7 @@ private void doTransferWrapper() throws IOException, WriteProcessException { private void doTransferWrapper(final Pair endPointAndBatch) throws IOException, WriteProcessException { final PipeTabletEventBatch batch = endPointAndBatch.getRight(); + logTransferBatch(endPointAndBatch.getLeft(), batch); if (batch instanceof PipeTabletEventPlainBatch) { doTransfer(endPointAndBatch.getLeft(), (PipeTabletEventPlainBatch) batch); } else if (batch instanceof PipeTabletEventTsFileBatch) { @@ -296,6 +298,27 @@ private void doTransferWrapper(final Pair endPo batch.onSuccess(); } + private void logTransferBatch(final TEndPoint endPoint, final PipeTabletEventBatch batch) { + if (!LOGGER.isDebugEnabled()) { + return; + } + + final List events = batch.deepCopyEvents(); + final EnrichedEvent firstEvent = events.isEmpty() ? null : events.get(0); + LOGGER.debug( + "{} sync sink emitting tablet batch, firstEventPipe={}, firstEventRegionId={}, endPoint={}, batchType={}, eventCount={}, batchBufferSize={}", + PipeDataLossDebugUtil.PREFIX, + firstEvent == null + ? "null" + : PipeDataLossDebugUtil.formatPipe( + firstEvent.getPipeName(), firstEvent.getCreationTime()), + firstEvent == null ? "null" : String.valueOf(firstEvent.getRegionId()), + endPoint, + batch.getClass().getSimpleName(), + batch.getEventCount(), + batch.getTotalBufferSize()); + } + private void doTransfer( final TEndPoint endPoint, final PipeTabletEventPlainBatch batchToTransfer) { final Pair clientAndStatus = clientManager.getClient(endPoint); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java index 3d81e143ac867..7605fb1c0397f 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java @@ -93,6 +93,9 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -156,6 +159,97 @@ public void testScanParser() throws Exception { System.out.println(System.currentTimeMillis() - startTime); } + @Test + public void testQueryParserKeepsPointsForHundredTimeSeriesSplitAlignedTsFile() throws Exception { + final int deviceCount = 10; + final int measurementCount = 100; + final int rowCount = 16; + + final List schemaList = new ArrayList<>(); + for (int measurementIndex = 0; measurementIndex < measurementCount; ++measurementIndex) { + schemaList.add(new MeasurementSchema("s" + measurementIndex, TSDataType.INT64)); + } + + alignedTsFile = new File("query-parser-100-timeseries-split-aligned.tsfile"); + if (alignedTsFile.exists()) { + Assert.assertTrue(alignedTsFile.delete()); + } + + try (final TsFileWriter writer = new TsFileWriter(alignedTsFile)) { + for (int deviceIndex = 0; deviceIndex < deviceCount; ++deviceIndex) { + final String device = "root.sg.d" + deviceIndex; + writer.registerAlignedTimeseries(new PartialPath(device), schemaList); + + final Tablet tablet = new Tablet(device, schemaList, rowCount); + for (int rowIndex = 0; rowIndex < rowCount; ++rowIndex) { + tablet.addTimestamp(rowIndex, rowIndex); + for (int measurementIndex = 0; measurementIndex < measurementCount; ++measurementIndex) { + tablet.addValue( + "s" + measurementIndex, + rowIndex, + (long) deviceIndex * measurementCount * rowCount + + (long) measurementIndex * rowCount + + rowIndex); + } + } + writer.writeAligned(tablet); + } + } + + int totalPointCount = 0; + int totalTabletCount = 0; + final ExecutorService executor = Executors.newFixedThreadPool(16); + try { + final List>> futures = new ArrayList<>(); + for (int measurementIndex = 0; measurementIndex < measurementCount; ++measurementIndex) { + final String measurement = "s" + measurementIndex; + futures.add( + executor.submit( + () -> { + int pointCount = 0; + int tabletCount = 0; + + final IoTDBTreePattern pattern = new IoTDBTreePattern("root.sg.*." + measurement); + Assert.assertFalse(pattern.mayMatchMultipleTimeSeriesInOneDevice()); + try (final TsFileInsertionEventQueryParser parser = + new TsFileInsertionEventQueryParser( + alignedTsFile, pattern, Long.MIN_VALUE, Long.MAX_VALUE, null)) { + final Iterator iterator = + parser.toTabletInsertionEvents().iterator(); + while (iterator.hasNext()) { + final PipeRawTabletInsertionEvent event = + (PipeRawTabletInsertionEvent) iterator.next(); + final Tablet parsedTablet = event.convertToTablet(); + + Assert.assertTrue(event.isAligned()); + Assert.assertEquals(1, parsedTablet.getSchemas().size()); + Assert.assertEquals( + measurement, parsedTablet.getSchemas().get(0).getMeasurementName()); + + pointCount += getNonNullSize(parsedTablet); + ++tabletCount; + } + } + + Assert.assertEquals(deviceCount * rowCount, pointCount); + Assert.assertTrue(tabletCount > 0); + return new Pair<>(pointCount, tabletCount); + })); + } + + for (final Future> future : futures) { + final Pair result = future.get(); + totalPointCount += result.getLeft(); + totalTabletCount += result.getRight(); + } + } finally { + executor.shutdownNow(); + } + + Assert.assertEquals(deviceCount * measurementCount * rowCount, totalPointCount); + Assert.assertTrue(totalTabletCount >= measurementCount); + } + @Test public void testScanParserReleasesTabletMemoryAfterRawTabletGenerated() throws Exception { nonalignedTsFile = diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java index 26e7ea305d595..47514f9921270 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java @@ -35,6 +35,7 @@ public class PipeEventCommitManager { private static final Logger LOGGER = LoggerFactory.getLogger(PipeEventCommitManager.class); + private static final String PIPE_DATA_LOSS_DEBUG_PREFIX = "[PipeDataLossDebug]"; private volatile PipeTaskAgent taskAgent; private final Map eventCommitterMap = new ConcurrentHashMap<>(); @@ -94,9 +95,23 @@ public void enrichWithCommitterKeyAndCommitId( return; } event.setCommitterKeyAndCommitId(committerKey, committer.generateCommitId()); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} commit manager enriched event, committerKey={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + committerKey, + event.coreReportMessage()); + } } public void commit(final EnrichedEvent event, final CommitterKey committerKey) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} commit manager commit invoked, committerKey={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + committerKey, + event == null ? null : event.coreReportMessage()); + } if (event == null || !event.needToCommit() || Objects.isNull(event.getPipeName()) @@ -118,6 +133,12 @@ public void commit(final EnrichedEvent event, final CommitterKey committerKey) { } } if (committerKey == null) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} commit manager skip commit because committerKey is null, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + event.coreReportMessage()); + } return; } if (event.hasMultipleCommitIds()) { @@ -138,6 +159,14 @@ private void commitMultipleIds(final CommitterKey committerKey, final EnrichedEv private boolean commitSingleId( final CommitterKey committerKey, final long commitId, final EnrichedEvent event) { if (commitId <= EnrichedEvent.NO_COMMIT_ID) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} commit manager skip commit because commitId is invalid, committerKey={}, commitId={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + committerKey, + commitId, + event.coreReportMessage()); + } return false; } final PipeEventCommitter committer = eventCommitterMap.get(committerKey); @@ -164,6 +193,14 @@ private boolean commitSingleId( return false; } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} commit manager submitting event to committer, committerKey={}, commitId={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + committerKey, + commitId, + event.coreReportMessage()); + } committer.commit(event); return true; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitter.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitter.java index 42429fca25365..3cfe86329ad21 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitter.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitter.java @@ -23,11 +23,17 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.pipe.api.event.Event; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.util.concurrent.atomic.AtomicLong; /** Used to queue {@link Event}s for one pipe of one region to commit in order. */ public class PipeEventCommitter { + private static final Logger LOGGER = LoggerFactory.getLogger(PipeEventCommitter.class); + private static final String PIPE_DATA_LOSS_DEBUG_PREFIX = "[PipeDataLossDebug]"; + private final CommitterKey committerKey; private final AtomicLong commitIdGenerator = new AtomicLong(0); @@ -40,17 +46,42 @@ public class PipeEventCommitter { } public synchronized long generateCommitId() { - return commitIdGenerator.incrementAndGet(); + final long commitId = commitIdGenerator.incrementAndGet(); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} committer generated commit id, committerKey={}, commitId={}, queueSize={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + committerKey, + commitId, + commitQueue.size()); + } + return commitId; } @SuppressWarnings("java:S899") public synchronized void commit(final EnrichedEvent event) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} committer commit start, committerKey={}, queueSizeBefore={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + committerKey, + commitQueue.size(), + event.coreReportMessage()); + } if (event.hasMultipleCommitIds()) { for (final EnrichedEvent dummyEvent : event.getDummyEventsForCommitIds()) { commitQueue.offer(dummyEvent); } } commitQueue.offer(event); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} committer commit end, committerKey={}, queueSizeAfter={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + committerKey, + commitQueue.size(), + event.coreReportMessage()); + } } //////////////////////////// APIs provided for metric framework //////////////////////////// diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/interval/PipeCommitInterval.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/interval/PipeCommitInterval.java index 456acd646dccf..ed227d3b5b82a 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/interval/PipeCommitInterval.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/interval/PipeCommitInterval.java @@ -24,12 +24,18 @@ import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; import org.apache.iotdb.commons.pipe.datastructure.interval.Interval; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.util.ArrayList; import java.util.List; import java.util.Objects; public class PipeCommitInterval extends Interval { + private static final Logger LOGGER = LoggerFactory.getLogger(PipeCommitInterval.class); + private static final String PIPE_DATA_LOSS_DEBUG_PREFIX = "[PipeDataLossDebug]"; + private ProgressIndex currentIndex; private List onCommittedHooks; private final PipeTaskMeta pipeTaskMeta; @@ -64,6 +70,16 @@ public void onMerged(final PipeCommitInterval another) { @Override public void onRemoved() { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} commit interval removed, range=[{},{}], progressIndex={}, hookCount={}, willReportProgress={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + start, + end, + currentIndex, + onCommittedHooks.size(), + Objects.nonNull(pipeTaskMeta)); + } if (Objects.nonNull(pipeTaskMeta)) { pipeTaskMeta.updateProgressIndex(currentIndex); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/interval/PipeCommitQueue.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/interval/PipeCommitQueue.java index 7b48aa39c9a2b..b99461b997140 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/interval/PipeCommitQueue.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/interval/PipeCommitQueue.java @@ -23,11 +23,25 @@ import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + public class PipeCommitQueue { + private static final Logger LOGGER = LoggerFactory.getLogger(PipeCommitQueue.class); + private static final String PIPE_DATA_LOSS_DEBUG_PREFIX = "[PipeDataLossDebug]"; + private final IntervalManager intervalManager = new IntervalManager<>(); private long lastCommitted = 0; public void offer(final EnrichedEvent event) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} commit queue offer start, lastCommitted={}, queueSizeBefore={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + lastCommitted, + intervalManager.size(), + event.coreReportMessage()); + } final PipeCommitInterval interval = new PipeCommitInterval( event.getCommitId(), @@ -39,9 +53,25 @@ public void offer(final EnrichedEvent event) { event.getPipeTaskMeta()); intervalManager.addInterval(interval); if (interval.start == lastCommitted + 1) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} commit queue interval is contiguous and will be removed, lastCommittedBefore={}, interval={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + lastCommitted, + interval, + event.coreReportMessage()); + } intervalManager.remove(interval); lastCommitted = interval.end; } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} commit queue offer end, lastCommitted={}, queueSizeAfter={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + lastCommitted, + intervalManager.size(), + event.coreReportMessage()); + } } public int size() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java index 63eedf80a43e5..115a04cc9e37e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java @@ -48,6 +48,7 @@ public abstract class EnrichedEvent implements Event { private static final Logger LOGGER = LoggerFactory.getLogger(EnrichedEvent.class); + private static final String PIPE_DATA_LOSS_DEBUG_PREFIX = "[PipeDataLossDebug]"; protected final AtomicInteger referenceCount; // This variable is used to indicate whether the event's reference count has ever been decreased @@ -133,6 +134,7 @@ protected void trackResource() { */ public synchronized boolean increaseReferenceCount(final String holderMessage) { boolean isSuccessful = true; + final int oldReferenceCount = referenceCount.get(); if (isReleased.get()) { LOGGER.warn( @@ -149,7 +151,17 @@ public synchronized boolean increaseReferenceCount(final String holderMessage) { } if (isSuccessful) { - if (referenceCount.incrementAndGet() == 1 + final int newReferenceCount = referenceCount.incrementAndGet(); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} enriched event increaseReferenceCount, holder={}, oldReferenceCount={}, newReferenceCount={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + holderMessage, + oldReferenceCount, + newReferenceCount, + coreReportMessage()); + } + if (newReferenceCount == 1 && PipeConfig.getInstance().getPipeEventReferenceTrackingEnabled()) { trackResource(); } @@ -190,6 +202,7 @@ public synchronized boolean increaseReferenceCount(final String holderMessage) { public synchronized boolean decreaseReferenceCount( final String holderMessage, final boolean shouldReport) { boolean isSuccessful = true; + final int oldReferenceCount = referenceCount.get(); if (isReleased.get()) { LOGGER.warn( @@ -205,6 +218,19 @@ public synchronized boolean decreaseReferenceCount( if (!shouldReport) { shouldReportOnCommit = false; } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} enriched event reference will reach zero, holder={}, shouldReportArg={}, shouldReportOnCommit={}, committerKey={}, commitId={}, commitIds={}, progressIndex={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + holderMessage, + shouldReport, + shouldReportOnCommit, + committerKey, + commitId, + getCommitIds(), + getProgressIndex(), + coreReportMessage()); + } // We assume that this function will not throw any exceptions. if (!internallyDecreaseResourceReferenceCount(holderMessage)) { LOGGER.warn( @@ -218,6 +244,16 @@ public synchronized boolean decreaseReferenceCount( // No matter whether the resource is released, we should decrease the reference count. final int newReferenceCount = referenceCount.decrementAndGet(); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} enriched event decreaseReferenceCount, holder={}, shouldReportArg={}, oldReferenceCount={}, newReferenceCount={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + holderMessage, + shouldReport, + oldReferenceCount, + newReferenceCount, + coreReportMessage()); + } if (newReferenceCount <= 0) { isReleased.set(true); isSuccessful = newReferenceCount == 0; @@ -253,9 +289,25 @@ public synchronized boolean decreaseReferenceCount( */ public synchronized boolean clearReferenceCount(final String holderMessage) { if (isReleased.get()) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} enriched event clearReferenceCount ignored because already released, holder={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + holderMessage, + coreReportMessage()); + } return false; } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} enriched event clearReferenceCount, holder={}, oldReferenceCount={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + holderMessage, + referenceCount.get(), + coreReportMessage()); + } + if (referenceCount.get() >= 1) { shouldReportOnCommit = false; // We assume that this function will not throw any exceptions. @@ -445,6 +497,14 @@ public void throwIfNoPrivilege() throws Exception { public void setCommitterKeyAndCommitId(final CommitterKey committerKey, final long commitId) { this.committerKey = committerKey; this.commitId = commitId; + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "{} enriched event bound committer key and commit id, committerKey={}, commitId={}, event={}", + PIPE_DATA_LOSS_DEBUG_PREFIX, + committerKey, + commitId, + coreReportMessage()); + } } public void setRebootTimes(final int rebootTimes) {