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 eec963793bd Fix aligned memtable flush after deleting measurements
(#18116)
eec963793bd is described below
commit eec963793bdf50d8dd9a0bb8a10583eb3fb5d3a4
Author: Caideyipi <[email protected]>
AuthorDate: Mon Jul 6 17:54:09 2026 +0800
Fix aligned memtable flush after deleting measurements (#18116)
---
.../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 58aced70ed3..c5d2e3b1781 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 class IoTDBPipeReceiverAutoCreateDisabledIT extends
AbstractPipeDualTreeM
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 949466a9f97..b0d00562a4c 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 class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
@Override
public IChunkWriter createIChunkWriter() {
- return new AlignedChunkWriterImpl(schemaList, encryptParameter);
+ return new AlignedChunkWriterImpl(getActiveSchemaList(), encryptParameter);
+ }
+
+ private List<IMeasurementSchema> getActiveSchemaList() {
+ if (measurementIndexMap.size() == schemaList.size()) {
+ return schemaList;
+ }
+ List<IMeasurementSchema> 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<TSDataType> getDataTypes(List<IMeasurementSchema> schemaList) {
+ if (schemaList == this.schemaList) {
+ return dataTypes;
+ }
+ List<TSDataType> 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 class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
BlockingQueue<Object> ioTaskQueue,
long maxNumberOfPointsInChunk,
int maxNumberOfPointsInPage) {
+ List<IMeasurementSchema> activeSchemaList = getActiveSchemaList();
+ if (activeSchemaList.isEmpty() && ignoreAllNullRows) {
+ return;
+ }
+
BitMap allValueColDeletedMap;
AlignedTVList alignedWorkingListForFlush = (AlignedTVList)
workingListForFlush;
@@ -615,7 +646,12 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
}
handleEncoding(
- ioTaskQueue, chunkRange, timeDuplicateInfo, allValueColDeletedMap,
maxNumberOfPointsInPage);
+ ioTaskQueue,
+ chunkRange,
+ timeDuplicateInfo,
+ allValueColDeletedMap,
+ maxNumberOfPointsInPage,
+ activeSchemaList);
}
private void handleEncoding(
@@ -623,21 +659,23 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
List<List<Integer>> chunkRange,
boolean[] timeDuplicateInfo,
BitMap allValueColDeletedMap,
- int maxNumberOfPointsInPage) {
+ int maxNumberOfPointsInPage,
+ List<IMeasurementSchema> activeSchemaList) {
AlignedTVList alignedWorkingListForFlush = (AlignedTVList)
workingListForFlush;
- List<TSDataType> dataTypes = alignedWorkingListForFlush.getTsDataTypes();
- Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new
Pair[dataTypes.size()];
+ List<Integer> columnIndexList = buildColumnIndexList(activeSchemaList);
+ Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new
Pair[activeSchemaList.size()];
for (List<Integer> 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 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
// 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 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
// 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 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
isNull
? 0
: alignedWorkingListForFlush.getIntByValueIndex(
- originRowIndex, columnIndex),
+ originRowIndex, tvListColumnIndex),
isNull);
break;
case INT64:
@@ -705,7 +748,7 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
isNull
? 0
: alignedWorkingListForFlush.getLongByValueIndex(
- originRowIndex, columnIndex),
+ originRowIndex, tvListColumnIndex),
isNull);
break;
case FLOAT:
@@ -714,7 +757,7 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
isNull
? 0
: alignedWorkingListForFlush.getFloatByValueIndex(
- originRowIndex, columnIndex),
+ originRowIndex, tvListColumnIndex),
isNull);
break;
case DOUBLE:
@@ -723,7 +766,7 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
isNull
? 0
: alignedWorkingListForFlush.getDoubleByValueIndex(
- originRowIndex, columnIndex),
+ originRowIndex, tvListColumnIndex),
isNull);
break;
case TEXT:
@@ -735,7 +778,7 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
isNull
? null
: alignedWorkingListForFlush.getBinaryByValueIndex(
- originRowIndex, columnIndex),
+ originRowIndex, tvListColumnIndex),
isNull);
break;
default:
@@ -787,16 +830,21 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
return;
}
+ List<IMeasurementSchema> activeSchemaList = getActiveSchemaList();
+ if (activeSchemaList.isEmpty() && ignoreAllNullRows) {
+ return;
+ }
+
AlignedChunkWriterImpl alignedChunkWriter =
- new AlignedChunkWriterImpl(schemaList, encryptParameter);
+ new AlignedChunkWriterImpl(activeSchemaList, encryptParameter);
// create MergeSortAlignedTVListIterator.
List<AlignedTVList> alignedTvLists = new ArrayList<>(sortedList);
alignedTvLists.add((AlignedTVList) workingListForFlush);
- List<Integer> columnIndexList = buildColumnIndexList(schemaList);
+ List<Integer> columnIndexList = buildColumnIndexList(activeSchemaList);
MemPointIterator timeValuePairIterator =
MemPointIteratorFactory.create(
- dataTypes,
+ getDataTypes(activeSchemaList),
columnIndexList,
alignedTvLists,
ignoreAllNullRows,
@@ -818,7 +866,7 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
- alignedChunkWriter = new AlignedChunkWriterImpl(schemaList,
encryptParameter);
+ alignedChunkWriter = new AlignedChunkWriterImpl(activeSchemaList,
encryptParameter);
encodeInfo.reset();
}
}
