This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch fix_tvlist_partial_clone_dev_1.3 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1086def0589033883e0cc3da9ac1e8e2bb891c09 Author: shuwenwei <[email protected]> AuthorDate: Wed Aug 12 16:42:36 2026 +0800 test: add aligned partial-column clone end-to-end test --- .../fragment/FragmentInstanceExecutionTest.java | 145 +++++++++++++++++++++ 1 file changed, 145 insertions(+) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java index 0f1b1c7d253..8bde6b302d5 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java @@ -22,6 +22,7 @@ package org.apache.iotdb.db.queryengine.execution.fragment; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.path.AlignedPath; import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -48,9 +49,12 @@ import com.google.common.collect.ImmutableSet; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.enums.CompressionType; import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.reader.IPointReader; +import org.apache.tsfile.utils.TsPrimitiveType; import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.BeforeClass; import org.junit.Test; import org.mockito.Mockito; @@ -58,7 +62,9 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.PrintStream; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; @@ -67,11 +73,23 @@ import static org.apache.iotdb.db.queryengine.common.QueryId.MOCK_QUERY_ID; import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; public class FragmentInstanceExecutionTest { + @BeforeClass + public static void setUpClass() { + // Initialize DataNodeId before any test to avoid ExceptionInInitializerError when + // Coordinator.<clinit> is triggered indirectly by async state-change listeners + // (e.g., via QueryRelatedResourceMetricSet -> Coordinator -> QueryIdGenerator). + IoTDBDescriptor.getInstance().getConfig().setDataNodeId(1); + } + @Test public void testFragmentInstanceExecution() { ExecutorService instanceNotificationExecutor = @@ -239,6 +257,133 @@ public class FragmentInstanceExecutionTest { } } + @Test + public void testAlignedTVListPartialColumnCloneEndToEnd() { + ExecutorService instanceNotificationExecutor = + IoTDBThreadPoolFactory.newFixedThreadPool(2, "test-aligned-partial-clone"); + try { + // Create an unsorted aligned MemTable with 5 columns. Row i (0..99) has + // value i * 100 + j in column j, and rows are inserted in reverse order. + List<IMeasurementSchema> schemaList = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + schemaList.add(new MeasurementSchema("sensor_" + i, TSDataType.INT64)); + } + String deviceId = "d1"; + IMemTable memTable = createMemTable(deviceId, schemaList); + + assertEquals(1, memTable.getMemTableMap().size()); + IWritableMemChunkGroup memChunkGroup = memTable.getMemTableMap().values().iterator().next(); + assertEquals(1, memChunkGroup.getMemChunkMap().size()); + IWritableMemChunk memChunk = memChunkGroup.getMemChunkMap().values().iterator().next(); + AlignedTVList workingTvList = (AlignedTVList) memChunk.getWorkingTVList(); + assertFalse(workingTvList.isSorted()); + assertEquals(100, workingTvList.rowCount()); + + // Two concurrent query contexts. + FragmentInstanceId id1 = new FragmentInstanceId(new PlanFragmentId(MOCK_QUERY_ID, 1), "1"); + FragmentInstanceStateMachine stateMachine1 = + new FragmentInstanceStateMachine(id1, instanceNotificationExecutor); + FragmentInstanceContext context1 = createFragmentInstanceContext(id1, stateMachine1); + + FragmentInstanceId id2 = new FragmentInstanceId(new PlanFragmentId(MOCK_QUERY_ID, 2), "2"); + FragmentInstanceStateMachine stateMachine2 = + new FragmentInstanceStateMachine(id2, instanceNotificationExecutor); + FragmentInstanceContext context2 = createFragmentInstanceContext(id2, stateMachine2); + + // Query 1: sensor_2 and sensor_0. It stays active on the unsorted working TVList + // (no point reader is opened yet, so the list is not sorted). + List<String> measurements1 = Arrays.asList("sensor_2", "sensor_0"); + List<IMeasurementSchema> schemas1 = Arrays.asList(schemaList.get(2), schemaList.get(0)); + AlignedPath fullPath1 = new AlignedPath(deviceId, measurements1, schemas1); + ReadOnlyMemChunk readOnlyMemChunk1 = + memTable.query(context1, fullPath1, Long.MIN_VALUE, null, null); + + // Query 2: sensor_1 and sensor_3. Because Query 1 is still active on the unsorted + // working TVList, this triggers clone-and-swap of the working TVList. + List<String> measurements2 = Arrays.asList("sensor_1", "sensor_3"); + List<IMeasurementSchema> schemas2 = Arrays.asList(schemaList.get(1), schemaList.get(3)); + AlignedPath fullPath2 = new AlignedPath(deviceId, measurements2, schemas2); + ReadOnlyMemChunk readOnlyMemChunk2 = + memTable.query(context2, fullPath2, Long.MIN_VALUE, null, null); + + // Query 1's columns (0 and 2) stay in the old working TVList; the other columns are + // moved to the clone. + assertEquals( + new HashSet<>(Arrays.asList(0, 2)), context1.getAccessedAlignedColumns(workingTvList)); + assertNotNull(workingTvList.getValues().get(0)); + assertNull(workingTvList.getValues().get(1)); + assertNotNull(workingTvList.getValues().get(2)); + assertNull(workingTvList.getValues().get(3)); + assertNull(workingTvList.getValues().get(4)); + + // The memChunk now points to the clone, which owns all 5 columns. + AlignedTVList cloneTvList = (AlignedTVList) memChunk.getWorkingTVList(); + assertNotSame(workingTvList, cloneTvList); + assertEquals(100, cloneTvList.rowCount()); + assertNotNull(cloneTvList.getValues().get(0)); + assertNotNull(cloneTvList.getValues().get(1)); + assertNotNull(cloneTvList.getValues().get(2)); + assertNotNull(cloneTvList.getValues().get(3)); + assertNotNull(cloneTvList.getValues().get(4)); + + // The old working TVList is owned by Query 1, which also reserved its memory. + assertSame(context1, workingTvList.getOwnerQuery()); + assertTrue(workingTvList.getReservedMemoryBytes() > 0); + + // Both queries must still read all 100 rows with correct values. + IPointReader pointReader1 = readOnlyMemChunk1.getPointReader(); + IPointReader pointReader2 = readOnlyMemChunk2.getPointReader(); + long row = 0; + while (pointReader1.hasNextTimeValuePair() && pointReader2.hasNextTimeValuePair()) { + TimeValuePair tvPair1 = pointReader1.nextTimeValuePair(); + TimeValuePair tvPair2 = pointReader2.nextTimeValuePair(); + + assertEquals(row, tvPair1.getTimestamp()); + assertEquals(row, tvPair2.getTimestamp()); + + // Query 1 reads [sensor_2, sensor_0] in query order. + TsPrimitiveType[] values1 = tvPair1.getValue().getVector(); + assertEquals(2, values1.length); + assertEquals(row * 100 + 2, values1[0].getLong()); + assertEquals(row * 100 + 0, values1[1].getLong()); + + // Query 2 reads [sensor_1, sensor_3] in query order. + TsPrimitiveType[] values2 = tvPair2.getValue().getVector(); + assertEquals(2, values2.length); + assertEquals(row * 100 + 1, values2[0].getLong()); + assertEquals(row * 100 + 3, values2[1].getLong()); + row++; + } + assertEquals(100, row); + assertFalse(pointReader1.hasNextTimeValuePair()); + assertFalse(pointReader2.hasNextTimeValuePair()); + + // Before ending the queries, the retained source's reserved memory must equal the value + // recalculated at cleanup time - this is exactly the accounting releaseTVListOwnedByQuery + // validates when the owning query ends (a mismatch would emit a WARN during release). + long reservedBeforeRelease = workingTvList.getReservedMemoryBytes(); + assertTrue(reservedBeforeRelease > 0); + assertEquals(workingTvList.calculateRamSize().getRamSize(), reservedBeforeRelease); + + // End Query 1, the owner of the retained source. No other query uses the source any more, + // so the final release must return its reserved memory and clear the list. + context1.releaseResource(); + assertTrue(workingTvList.getQueryContextSet().isEmpty()); + assertEquals(0, workingTvList.rowCount()); + + // End Query 2. The clone is now the memTable's working TVList, so cleanup only detaches the + // query from it; the clone itself must remain intact as the working list. + context2.releaseResource(); + assertTrue(cloneTvList.getQueryContextSet().isEmpty()); + assertEquals(100, cloneTvList.rowCount()); + assertSame(cloneTvList, memChunk.getWorkingTVList()); + } catch (Exception e) { + fail(e.getMessage()); + } finally { + instanceNotificationExecutor.shutdown(); + } + } + private FragmentInstanceExecution createFragmentInstanceExecution(int id, Executor executor) throws CpuNotEnoughException { IDriverScheduler scheduler = Mockito.mock(IDriverScheduler.class);
