This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 449946d5268 Fix pipe event accounting and aligned tablet failure
reporting (#18150)
449946d5268 is described below
commit 449946d5268b949d441716122266c09b3f3b6dc7
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 9 14:44:28 2026 +0800
Fix pipe event accounting and aligned tablet failure reporting (#18150)
* Fix pollLast event counter update
* Pipe: release replaced progress report events
* Write: preserve aligned tablet row failures
---
.../dataregion/memtable/TsFileProcessor.java | 32 ++----
.../realtime/PipeRealtimeDataRegionSourceTest.java | 113 +++++++++++++++++++++
.../dataregion/memtable/TsFileProcessorTest.java | 74 ++++++++++++++
.../connection/UnboundedBlockingPendingQueue.java | 4 +-
.../UnboundedBlockingPendingQueueTest.java | 73 +++++++++++++
5 files changed, 273 insertions(+), 23 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 680dfd9d0e1..6dc5ecb3e2b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -491,7 +491,6 @@ public class TsFileProcessor {
InsertTabletNode insertTabletNode,
List<int[]> rangeList,
TSStatus[] results,
- boolean noFailure,
long[] infoForMetrics)
throws WriteProcessException {
long memControlStartTime = System.nanoTime();
@@ -500,7 +499,7 @@ public class TsFileProcessor {
int start = range[0];
int end = range[1];
try {
- long[] memIncrements = checkMemCost(insertTabletNode, start, end,
noFailure, results);
+ long[] memIncrements = checkMemCost(insertTabletNode, start, end,
results);
for (int i = 0; i < memIncrements.length; i++) {
totalMemIncrements[i] += memIncrements[i];
}
@@ -518,11 +517,11 @@ public class TsFileProcessor {
}
private long[] checkMemCost(
- InsertTabletNode insertTabletNode, int start, int end, boolean
noFailure, TSStatus[] results)
+ InsertTabletNode insertTabletNode, int start, int end, TSStatus[]
results)
throws WriteProcessException {
long[] memIncrements;
if (insertTabletNode.isAligned()) {
- memIncrements = checkAlignedMemCost(insertTabletNode, start, end,
noFailure, results);
+ memIncrements = checkAlignedMemCost(insertTabletNode, start, end,
results);
} else {
memIncrements =
checkMemCostAndAddToTspInfoForTablet(
@@ -538,7 +537,7 @@ public class TsFileProcessor {
}
private long[] checkAlignedMemCost(
- InsertTabletNode insertTabletNode, int start, int end, boolean
noFailure, TSStatus[] results)
+ InsertTabletNode insertTabletNode, int start, int end, TSStatus[]
results)
throws WriteProcessException {
List<Pair<IDeviceID, Integer>> deviceEndPosList =
insertTabletNode.splitByDevice(start, end);
long[] memIncrements = new long[NUM_MEM_TO_ESTIMATE];
@@ -555,7 +554,6 @@ public class TsFileProcessor {
insertTabletNode.getColumnCategories(),
splitStart,
splitEnd,
- noFailure,
results);
for (int i = 0; i < NUM_MEM_TO_ESTIMATE; i++) {
memIncrements[i] += splitMemIncrements[i];
@@ -586,7 +584,7 @@ public class TsFileProcessor {
workMemTable.checkDataType(insertTabletNode);
long[] memIncrements =
- scheduleMemoryBlock(insertTabletNode, rangeList, results, noFailure,
infoForMetrics);
+ scheduleMemoryBlock(insertTabletNode, rangeList, results,
infoForMetrics);
long startTime = System.nanoTime();
WALFlushListener walFlushListener;
@@ -648,7 +646,10 @@ public class TsFileProcessor {
throw new WriteProcessException(e);
}
for (int i = start; i < end; i++) {
- results[i] = RpcUtils.SUCCESS_STATUS;
+ if (results[i] == null
+ || results[i].getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ results[i] = RpcUtils.SUCCESS_STATUS;
+ }
}
final List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs =
@@ -977,7 +978,6 @@ public class TsFileProcessor {
TsTableColumnCategory[] columnCategories,
int start,
int end,
- boolean noFailure,
TSStatus[] results)
throws WriteProcessException {
if (start >= end) {
@@ -994,7 +994,6 @@ public class TsFileProcessor {
memIncrements,
columns,
columnCategories,
- noFailure,
results);
long memTableIncrement = memIncrements[0];
long textDataIncrement = memIncrements[1];
@@ -1051,19 +1050,8 @@ public class TsFileProcessor {
long[] memIncrements,
Object[] columns,
TsTableColumnCategory[] columnCategories,
- boolean noFailure,
TSStatus[] results) {
- int incomingPointNum;
- if (noFailure) {
- incomingPointNum = end - start;
- } else {
- incomingPointNum = end - start;
- for (TSStatus result : results) {
- if (result != null && result.code !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- incomingPointNum--;
- }
- }
- }
+ int incomingPointNum = end - start;
TSDataType[] writableFieldDataTypes =
getWritableFieldDataTypes(measurementIds, dataTypes, columns,
columnCategories);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSourceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSourceTest.java
new file mode 100644
index 00000000000..4677f2126b6
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSourceTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.source.dataregion.realtime;
+
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
+import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEventFactory;
+import org.apache.iotdb.pipe.api.event.Event;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class PipeRealtimeDataRegionSourceTest {
+
+ private static final String TEST_REFERENCE_HOLDER =
+ PipeRealtimeDataRegionSourceTest.class.getName();
+
+ @Test
+ public void progressReportEventReleasesDroppedHeartbeatEvent() throws
Exception {
+ try (final ProgressReportTestSource source = new
ProgressReportTestSource()) {
+ final PipeRealtimeEvent heartbeatEvent = createHeartbeatEvent();
+ source.extract(heartbeatEvent);
+ Assert.assertEquals(1, source.getEventCount());
+ Assert.assertEquals(1, source.getPipeHeartbeatEventCount());
+
+ final PipeRealtimeEvent progressReportEvent =
createProgressReportEvent();
+ source.extract(progressReportEvent);
+
+ Assert.assertTrue(heartbeatEvent.getEvent().isReleased());
+ Assert.assertEquals(0, heartbeatEvent.getEvent().getReferenceCount());
+ Assert.assertEquals(1, source.getEventCount());
+ Assert.assertEquals(0, source.getPipeHeartbeatEventCount());
+ Assert.assertFalse(progressReportEvent.getEvent().isReleased());
+ }
+ }
+
+ @Test
+ public void mergedProgressReportEventReleasesNewEvent() throws Exception {
+ try (final ProgressReportTestSource source = new
ProgressReportTestSource()) {
+ final PipeRealtimeEvent firstProgressReportEvent =
createProgressReportEvent();
+ final PipeRealtimeEvent secondProgressReportEvent =
createProgressReportEvent();
+
+ source.extract(firstProgressReportEvent);
+ source.extract(secondProgressReportEvent);
+
+ Assert.assertFalse(firstProgressReportEvent.getEvent().isReleased());
+ Assert.assertTrue(secondProgressReportEvent.getEvent().isReleased());
+ Assert.assertEquals(1, source.getEventCount());
+ }
+ }
+
+ private static PipeRealtimeEvent createHeartbeatEvent() {
+ final PipeRealtimeEvent event =
PipeRealtimeEventFactory.createRealtimeEvent(1, false);
+ Assert.assertTrue(event.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ return event;
+ }
+
+ private static PipeRealtimeEvent createProgressReportEvent() {
+ final ProgressReportEvent progressReportEvent = new
ProgressReportEvent("pipe", 1L, null);
+ progressReportEvent.bindProgressIndex(MinimumProgressIndex.INSTANCE);
+
+ final PipeRealtimeEvent event =
+ PipeRealtimeEventFactory.createRealtimeEvent(progressReportEvent);
+ Assert.assertTrue(event.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ return event;
+ }
+
+ private static class ProgressReportTestSource extends
PipeRealtimeDataRegionSource {
+
+ @Override
+ protected void doExtract(final PipeRealtimeEvent event) {
+ if (event.getEvent() instanceof PipeHeartbeatEvent) {
+ extractHeartbeat(event);
+ return;
+ }
+ pendingQueue.offer(event);
+ }
+
+ @Override
+ public Event supply() {
+ return pendingQueue.directPoll();
+ }
+
+ @Override
+ public boolean isNeedListenToTsFile() {
+ return false;
+ }
+
+ @Override
+ public boolean isNeedListenToInsertNode() {
+ return false;
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
index 2609cb1acb5..2b14116ff2d 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
@@ -41,9 +41,12 @@ import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalIn
import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo;
import org.apache.iotdb.db.storageengine.dataregion.DataRegionTest;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager;
import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.db.utils.constant.TestConstant;
+import org.apache.iotdb.rpc.RpcUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.external.commons.io.FileUtils;
@@ -73,6 +76,7 @@ import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -560,6 +564,33 @@ public class TsFileProcessorTest {
Assert.assertEquals(721560, memTable.memSize());
}
+ @Test
+ public void alignedTabletKeepsFailedStatusesAndCountsWrittenRows()
+ throws MetadataException, WriteProcessException, IOException,
IllegalPathException {
+ final int rowCount = PrimitiveArrayManager.ARRAY_SIZE + 2;
+ final List<int[]> rangeList = Collections.singletonList(new int[] {0,
rowCount - 1});
+
+ final TsFileProcessor expectedProcessor = newTestProcessor(filePath +
".expected");
+ final TSStatus[] expectedResults = new TSStatus[rowCount];
+ Arrays.fill(expectedResults, RpcUtils.SUCCESS_STATUS);
+ expectedProcessor.insertTablet(
+ genSingleMeasurementTablet(rowCount, true), rangeList,
expectedResults, false, new long[5]);
+
+ final TsFileProcessor actualProcessor = newTestProcessor(filePath +
".actual");
+ final TSStatus[] actualResults = new TSStatus[rowCount];
+ Arrays.fill(actualResults, RpcUtils.SUCCESS_STATUS);
+ final int failedIndex = rowCount - 2;
+ actualResults[failedIndex] = RpcUtils.getStatus(TSStatusCode.OUT_OF_TTL,
"failed row");
+ actualProcessor.insertTablet(
+ genSingleMeasurementTablet(rowCount, true), rangeList, actualResults,
false, new long[5]);
+
+ Assert.assertEquals(
+ expectedProcessor.getWorkMemTable().getTVListsRamCost(),
+ actualProcessor.getWorkMemTable().getTVListsRamCost());
+ Assert.assertEquals(
+ TSStatusCode.OUT_OF_TTL.getStatusCode(),
actualResults[failedIndex].getCode());
+ }
+
@Test
public void alignedTvListRamCostTest2()
throws MetadataException, WriteProcessException, IOException {
@@ -1261,6 +1292,49 @@ public class TsFileProcessorTest {
}
}
+ private TsFileProcessor newTestProcessor(String path) throws IOException,
WriteProcessException {
+ TsFileProcessor newProcessor =
+ new TsFileProcessor(
+ storageGroup,
+ SystemFileFactory.INSTANCE.getFile(path),
+ sgInfo,
+ this::closeTsFileProcessor,
+ (tsFileProcessor, updateMap, systemFlushTime) -> {},
+ true);
+ TsFileProcessorInfo tsFileProcessorInfo = new TsFileProcessorInfo(sgInfo);
+ newProcessor.setTsFileProcessorInfo(tsFileProcessorInfo);
+ this.sgInfo.initTsFileProcessorInfo(newProcessor);
+ SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, newProcessor);
+ return newProcessor;
+ }
+
+ private InsertTabletNode genSingleMeasurementTablet(int rowCount, boolean
isAligned)
+ throws IllegalPathException {
+ String[] measurements = new String[] {measurementId};
+ TSDataType[] dataTypes = new TSDataType[] {dataType};
+ MeasurementSchema[] schemas =
+ new MeasurementSchema[] {new MeasurementSchema(measurementId,
dataType, encoding)};
+ long[] times = new long[rowCount];
+ Object[] columns = new Object[] {new int[rowCount]};
+
+ for (int i = 0; i < rowCount; i++) {
+ times[i] = i;
+ ((int[]) columns[0])[i] = i;
+ }
+
+ return new InsertTabletNode(
+ new QueryId("test_write").genPlanNodeId(),
+ new PartialPath(deviceId),
+ isAligned,
+ measurements,
+ dataTypes,
+ schemas,
+ times,
+ null,
+ columns,
+ rowCount);
+ }
+
private InsertTabletNode genInsertTableNode(long startTime, boolean
isAligned)
throws IllegalPathException {
String deviceId = "root.sg.device5";
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java
index 43fa64c158e..ceae0180a36 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java
@@ -39,6 +39,8 @@ public class UnboundedBlockingPendingQueue<E extends Event>
extends BlockingPend
}
public E pollLast() {
- return pendingDeque.pollLast();
+ final E event = pendingDeque.pollLast();
+ eventCounter.decreaseEventCount(event);
+ return event;
}
}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueueTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueueTest.java
new file mode 100644
index 00000000000..ed2e7a214d9
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueueTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.commons.pipe.agent.task.connection;
+
+import org.apache.iotdb.commons.pipe.metric.PipeEventCounter;
+import org.apache.iotdb.pipe.api.event.Event;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class UnboundedBlockingPendingQueueTest {
+
+ @Test
+ public void pollLastDecreasesEventCount() {
+ final CountingEventCounter eventCounter = new CountingEventCounter();
+ final UnboundedBlockingPendingQueue<Event> queue =
+ new UnboundedBlockingPendingQueue<>(eventCounter);
+ final Event event = new Event() {};
+
+ queue.offer(event);
+ Assert.assertEquals(1, eventCounter.getEventCount());
+
+ Assert.assertSame(event, queue.pollLast());
+ Assert.assertEquals(0, eventCounter.getEventCount());
+ }
+
+ private static class CountingEventCounter extends PipeEventCounter {
+
+ private final AtomicInteger eventCount = new AtomicInteger();
+
+ private int getEventCount() {
+ return eventCount.get();
+ }
+
+ @Override
+ public void increaseEventCount(final Event event) {
+ if (event != null) {
+ eventCount.incrementAndGet();
+ }
+ }
+
+ @Override
+ public void decreaseEventCount(final Event event) {
+ if (event != null) {
+ eventCount.decrementAndGet();
+ }
+ }
+
+ @Override
+ public void reset() {
+ eventCount.set(0);
+ }
+ }
+}
