From 959f153145cd6e84cb13ef6d5c2e3cc210e9090e Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 6 Jul 2026 14:47:41 +0800 Subject: [PATCH] Fix aligned memtable flush after deleting measurements --- ...IoTDBPipeReceiverAutoCreateDisabledIT.java | 37 ++++++++ .../memtable/AlignedWritableMemChunk.java | 92 ++++++++++++++----- 2 files changed, 107 insertions(+), 22 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeReceiverAutoCreateDisabledIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeReceiverAutoCreateDisabledIT.java index 58aced70ed392..c5d2e3b178195 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeReceiverAutoCreateDisabledIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeReceiverAutoCreateDisabledIT.java @@ -182,6 +182,43 @@ public void testAutoSplitHistoryTsFileWithDeletionWhenReceiverAutoCreateSchemaDi new HashSet<>(Arrays.asList("root.sg.aligned,true,null,INF,"))); } + @Test + public void + testAutoSplitHistoryTsFileWithUnflushedAlignedDeletionWhenReceiverAutoCreateSchemaDisabled() + throws Exception { + TestUtils.executeNonQueries( + senderEnv, + Arrays.asList( + "create database root.sg", + "create aligned timeseries root.sg.aligned(s0 INT32, s1 INT64, s2 DOUBLE)", + "insert into root.sg.aligned(time, s0, s1, s2) aligned " + + "values(1, 1, 10, 1.0), (2, 2, 20, 2.0), (3, 3, 30, 3.0)", + "delete timeseries root.sg.aligned.s1")); + + TestUtils.executeNonQuery( + senderEnv, + String.format( + "create pipe test with source ('inclusion'='all', 'source.history.enable'='true', 'source.realtime.mode'='batch') " + + "with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')", + receiverEnv.getDataNodeWrapper(0).getIpAndPortString())); + + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select s0,s2 from root.sg.aligned", + "Time,root.sg.aligned.s0,root.sg.aligned.s2,", + new HashSet<>(Arrays.asList("1,1,1.0,", "2,2,2.0,", "3,3,3.0,"))); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "count timeseries root.sg.aligned.*", + "count(timeseries),", + new HashSet<>(Arrays.asList("2,"))); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "show devices root.sg.aligned", + "Device,IsAligned,Template,TTL(ms),", + new HashSet<>(Arrays.asList("root.sg.aligned,true,null,INF,"))); + } + private QueryResult queryForResult(final Statement statement, final String sql) throws SQLException { try (final ResultSet resultSet = statement.executeQuery(sql)) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java index 949466a9f979f..b0d00562a4c75 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java @@ -542,7 +542,33 @@ public void removeColumn(String measurementId) { @Override public IChunkWriter createIChunkWriter() { - return new AlignedChunkWriterImpl(schemaList, encryptParameter); + return new AlignedChunkWriterImpl(getActiveSchemaList(), encryptParameter); + } + + private List getActiveSchemaList() { + if (measurementIndexMap.size() == schemaList.size()) { + return schemaList; + } + List activeSchemaList = new ArrayList<>(measurementIndexMap.size()); + for (int i = 0; i < schemaList.size(); i++) { + IMeasurementSchema schema = schemaList.get(i); + Integer activeColumnIndex = measurementIndexMap.get(schema.getMeasurementName()); + if (activeColumnIndex != null && activeColumnIndex == i) { + activeSchemaList.add(schema); + } + } + return activeSchemaList; + } + + private List getDataTypes(List schemaList) { + if (schemaList == this.schemaList) { + return dataTypes; + } + List activeDataTypes = new ArrayList<>(schemaList.size()); + for (IMeasurementSchema schema : schemaList) { + activeDataTypes.add(schema.getType()); + } + return activeDataTypes; } @SuppressWarnings({"squid:S6541", "squid:S3776"}) @@ -550,6 +576,11 @@ public void encodeWorkingAlignedTVList( BlockingQueue ioTaskQueue, long maxNumberOfPointsInChunk, int maxNumberOfPointsInPage) { + List activeSchemaList = getActiveSchemaList(); + if (activeSchemaList.isEmpty() && ignoreAllNullRows) { + return; + } + BitMap allValueColDeletedMap; AlignedTVList alignedWorkingListForFlush = (AlignedTVList) workingListForFlush; @@ -615,7 +646,12 @@ public void encodeWorkingAlignedTVList( } handleEncoding( - ioTaskQueue, chunkRange, timeDuplicateInfo, allValueColDeletedMap, maxNumberOfPointsInPage); + ioTaskQueue, + chunkRange, + timeDuplicateInfo, + allValueColDeletedMap, + maxNumberOfPointsInPage, + activeSchemaList); } private void handleEncoding( @@ -623,21 +659,23 @@ private void handleEncoding( List> chunkRange, boolean[] timeDuplicateInfo, BitMap allValueColDeletedMap, - int maxNumberOfPointsInPage) { + int maxNumberOfPointsInPage, + List activeSchemaList) { AlignedTVList alignedWorkingListForFlush = (AlignedTVList) workingListForFlush; - List dataTypes = alignedWorkingListForFlush.getTsDataTypes(); - Pair[] lastValidPointIndexForTimeDupCheck = new Pair[dataTypes.size()]; + List columnIndexList = buildColumnIndexList(activeSchemaList); + Pair[] lastValidPointIndexForTimeDupCheck = new Pair[activeSchemaList.size()]; for (List pageRange : chunkRange) { AlignedChunkWriterImpl alignedChunkWriter = - new AlignedChunkWriterImpl(schemaList, encryptParameter); + new AlignedChunkWriterImpl(activeSchemaList, encryptParameter); for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum += 1) { - for (int columnIndex = 0; columnIndex < dataTypes.size(); columnIndex++) { + for (int columnIndex = 0; columnIndex < activeSchemaList.size(); columnIndex++) { + int tvListColumnIndex = columnIndexList.get(columnIndex); // Pair of Time and Index if (Objects.nonNull(timeDuplicateInfo) && lastValidPointIndexForTimeDupCheck[columnIndex] == null) { lastValidPointIndexForTimeDupCheck[columnIndex] = new Pair<>(Long.MIN_VALUE, null); } - TSDataType tsDataType = dataTypes.get(columnIndex); + TSDataType tsDataType = activeSchemaList.get(columnIndex).getType(); for (int sortedRowIndex = pageRange.get(pageNum * 2); sortedRowIndex <= pageRange.get(pageNum * 2 + 1); sortedRowIndex++) { @@ -650,8 +688,10 @@ private void handleEncoding( // skip time duplicated rows long time = alignedWorkingListForFlush.getTime(sortedRowIndex); if (Objects.nonNull(timeDuplicateInfo)) { - if (!alignedWorkingListForFlush.isNullValue( - alignedWorkingListForFlush.getValueIndex(sortedRowIndex), columnIndex)) { + if (tvListColumnIndex >= 0 + && !alignedWorkingListForFlush.isNullValue( + alignedWorkingListForFlush.getValueIndex(sortedRowIndex), + tvListColumnIndex)) { lastValidPointIndexForTimeDupCheck[columnIndex].left = time; lastValidPointIndexForTimeDupCheck[columnIndex].right = alignedWorkingListForFlush.getValueIndex(sortedRowIndex); @@ -671,21 +711,24 @@ private void handleEncoding( // write(T:3,V:null) int originRowIndex; - if (Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex]) + if (tvListColumnIndex >= 0 + && Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex]) && (time == lastValidPointIndexForTimeDupCheck[columnIndex].left)) { originRowIndex = lastValidPointIndexForTimeDupCheck[columnIndex].right; } else { originRowIndex = alignedWorkingListForFlush.getValueIndex(sortedRowIndex); } - boolean isNull = alignedWorkingListForFlush.isNullValue(originRowIndex, columnIndex); + boolean isNull = + tvListColumnIndex < 0 + || alignedWorkingListForFlush.isNullValue(originRowIndex, tvListColumnIndex); switch (tsDataType) { case BOOLEAN: alignedChunkWriter.writeByColumn( time, !isNull && alignedWorkingListForFlush.getBooleanByValueIndex( - originRowIndex, columnIndex), + originRowIndex, tvListColumnIndex), isNull); break; case INT32: @@ -695,7 +738,7 @@ private void handleEncoding( isNull ? 0 : alignedWorkingListForFlush.getIntByValueIndex( - originRowIndex, columnIndex), + originRowIndex, tvListColumnIndex), isNull); break; case INT64: @@ -705,7 +748,7 @@ private void handleEncoding( isNull ? 0 : alignedWorkingListForFlush.getLongByValueIndex( - originRowIndex, columnIndex), + originRowIndex, tvListColumnIndex), isNull); break; case FLOAT: @@ -714,7 +757,7 @@ private void handleEncoding( isNull ? 0 : alignedWorkingListForFlush.getFloatByValueIndex( - originRowIndex, columnIndex), + originRowIndex, tvListColumnIndex), isNull); break; case DOUBLE: @@ -723,7 +766,7 @@ private void handleEncoding( isNull ? 0 : alignedWorkingListForFlush.getDoubleByValueIndex( - originRowIndex, columnIndex), + originRowIndex, tvListColumnIndex), isNull); break; case TEXT: @@ -735,7 +778,7 @@ private void handleEncoding( isNull ? null : alignedWorkingListForFlush.getBinaryByValueIndex( - originRowIndex, columnIndex), + originRowIndex, tvListColumnIndex), isNull); break; default: @@ -787,16 +830,21 @@ public void encode(BlockingQueue ioTaskQueue, BatchEncodeInfo encodeInfo return; } + List activeSchemaList = getActiveSchemaList(); + if (activeSchemaList.isEmpty() && ignoreAllNullRows) { + return; + } + AlignedChunkWriterImpl alignedChunkWriter = - new AlignedChunkWriterImpl(schemaList, encryptParameter); + new AlignedChunkWriterImpl(activeSchemaList, encryptParameter); // create MergeSortAlignedTVListIterator. List alignedTvLists = new ArrayList<>(sortedList); alignedTvLists.add((AlignedTVList) workingListForFlush); - List columnIndexList = buildColumnIndexList(schemaList); + List columnIndexList = buildColumnIndexList(activeSchemaList); MemPointIterator timeValuePairIterator = MemPointIteratorFactory.create( - dataTypes, + getDataTypes(activeSchemaList), columnIndexList, alignedTvLists, ignoreAllNullRows, @@ -818,7 +866,7 @@ public void encode(BlockingQueue ioTaskQueue, BatchEncodeInfo encodeInfo } catch (InterruptedException e) { Thread.currentThread().interrupt(); } - alignedChunkWriter = new AlignedChunkWriterImpl(schemaList, encryptParameter); + alignedChunkWriter = new AlignedChunkWriterImpl(activeSchemaList, encryptParameter); encodeInfo.reset(); } }