From dbafc25f38f5b393c6f9e60db1f62820e67cc580 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 9 Jul 2026 14:44:28 +0800 Subject: [PATCH] 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 (cherry picked from commit 449946d5268b949d441716122266c09b3f3b6dc7) --- .../dataregion/memtable/TsFileProcessor.java | 5 +- .../PipeRealtimeDataRegionSourceTest.java | 113 ++++++++++++++++++ .../memtable/TsFileProcessorTest.java | 73 +++++++++++ .../UnboundedBlockingPendingQueue.java | 4 +- .../UnboundedBlockingPendingQueueTest.java | 73 +++++++++++ 5 files changed, 266 insertions(+), 2 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSourceTest.java create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueueTest.java 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 6bbcb427f8798..38b681c91deb7 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 @@ -560,7 +560,10 @@ public void insertTablet( 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; + } } tsFileResource.updateStartTime( 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 0000000000000..99114bc8d4397 --- /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 9a40ee0b36cc4..f1ab05283b815 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 @@ -39,9 +39,12 @@ 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.commons.io.FileUtils; import org.apache.tsfile.enums.TSDataType; @@ -72,6 +75,7 @@ import java.io.File; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; @@ -704,6 +708,32 @@ public void alignedTvListRamCostTest() Assert.assertEquals(721560, memTable.memSize()); } + @Test + public void alignedTabletKeepsFailedStatusesAndCountsWrittenRows() + throws MetadataException, WriteProcessException, IOException, IllegalPathException { + final int rowCount = PrimitiveArrayManager.ARRAY_SIZE + 2; + + final TsFileProcessor expectedProcessor = newTestProcessor(filePath + ".expected"); + final TSStatus[] expectedResults = new TSStatus[rowCount]; + Arrays.fill(expectedResults, RpcUtils.SUCCESS_STATUS); + expectedProcessor.insertTablet( + genSingleMeasurementTablet(rowCount, true), 0, rowCount - 1, expectedResults); + + 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), 0, rowCount - 1, actualResults); + + 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 { @@ -1183,6 +1213,49 @@ private void closeTsFileProcessor(TsFileProcessor unsealedTsFileProcessor) } } + 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 43fa64c158ea4..ceae0180a36c4 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 E peekLast() { } 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 0000000000000..ed2e7a214d9f5 --- /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 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); + } + } +}