Skip to content

Commit 271906d

Browse files
CaideyipiCaideyipi
committed
Fix aligned memtable flush after deleting measurements
1 parent 8940ba0 commit 271906d

2 files changed

Lines changed: 101 additions & 22 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeReceiverAutoCreateDisabledIT.java‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,43 @@ public void testAutoSplitHistoryTsFileWithDeletionWhenReceiverAutoCreateSchemaDi
182182
new HashSet<>(Arrays.asList("root.sg.aligned,true,null,INF,")));
183183
}
184184

185+
@Test
186+
public void
187+
testAutoSplitHistoryTsFileWithUnflushedAlignedDeletionWhenReceiverAutoCreateSchemaDisabled()
188+
throws Exception {
189+
TestUtils.executeNonQueries(
190+
senderEnv,
191+
Arrays.asList(
192+
"create database root.sg",
193+
"create aligned timeseries root.sg.aligned(s0 INT32, s1 INT64, s2 DOUBLE)",
194+
"insert into root.sg.aligned(time, s0, s1, s2) aligned "
195+
+ "values(1, 1, 10, 1.0), (2, 2, 20, 2.0), (3, 3, 30, 3.0)",
196+
"delete timeseries root.sg.aligned.s1"));
197+
198+
TestUtils.executeNonQuery(
199+
senderEnv,
200+
String.format(
201+
"create pipe test with source ('inclusion'='all', 'source.history.enable'='true', 'source.realtime.mode'='batch') "
202+
+ "with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
203+
receiverEnv.getDataNodeWrapper(0).getIpAndPortString()));
204+
205+
TestUtils.assertDataEventuallyOnEnv(
206+
receiverEnv,
207+
"select s0,s2 from root.sg.aligned",
208+
"Time,root.sg.aligned.s0,root.sg.aligned.s2,",
209+
new HashSet<>(Arrays.asList("1,1,1.0,", "2,2,2.0,", "3,3,3.0,")));
210+
TestUtils.assertDataEventuallyOnEnv(
211+
receiverEnv,
212+
"count timeseries root.sg.aligned.*",
213+
"count(timeseries),",
214+
new HashSet<>(Arrays.asList("2,")));
215+
TestUtils.assertDataEventuallyOnEnv(
216+
receiverEnv,
217+
"show devices root.sg.aligned",
218+
"Device,IsAligned,Template,TTL(ms),",
219+
new HashSet<>(Arrays.asList("root.sg.aligned,true,null,INF,")));
220+
}
221+
185222
private QueryResult queryForResult(final Statement statement, final String sql)
186223
throws SQLException {
187224
try (final ResultSet resultSet = statement.executeQuery(sql)) {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java‎

Lines changed: 64 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -542,14 +542,39 @@ public void removeColumn(String measurementId) {
542542

543543
@Override
544544
public IChunkWriter createIChunkWriter() {
545-
return new AlignedChunkWriterImpl(schemaList, encryptParameter);
545+
return new AlignedChunkWriterImpl(getActiveSchemaList(), encryptParameter);
546+
}
547+
548+
private List<IMeasurementSchema> getActiveSchemaList() {
549+
List<IMeasurementSchema> activeSchemaList = new ArrayList<>(measurementIndexMap.size());
550+
for (int i = 0; i < schemaList.size(); i++) {
551+
IMeasurementSchema schema = schemaList.get(i);
552+
Integer activeColumnIndex = measurementIndexMap.get(schema.getMeasurementName());
553+
if (activeColumnIndex != null && activeColumnIndex == i) {
554+
activeSchemaList.add(schema);
555+
}
556+
}
557+
return activeSchemaList;
558+
}
559+
560+
private List<TSDataType> getDataTypes(List<IMeasurementSchema> schemaList) {
561+
List<TSDataType> activeDataTypes = new ArrayList<>(schemaList.size());
562+
for (IMeasurementSchema schema : schemaList) {
563+
activeDataTypes.add(schema.getType());
564+
}
565+
return activeDataTypes;
546566
}
547567

548568
@SuppressWarnings({"squid:S6541", "squid:S3776"})
549569
public void encodeWorkingAlignedTVList(
550570
BlockingQueue<Object> ioTaskQueue,
551571
long maxNumberOfPointsInChunk,
552572
int maxNumberOfPointsInPage) {
573+
List<IMeasurementSchema> activeSchemaList = getActiveSchemaList();
574+
if (activeSchemaList.isEmpty()) {
575+
return;
576+
}
577+
553578
BitMap allValueColDeletedMap;
554579
AlignedTVList alignedWorkingListForFlush = (AlignedTVList) workingListForFlush;
555580

@@ -615,29 +640,36 @@ public void encodeWorkingAlignedTVList(
615640
}
616641

617642
handleEncoding(
618-
ioTaskQueue, chunkRange, timeDuplicateInfo, allValueColDeletedMap, maxNumberOfPointsInPage);
643+
ioTaskQueue,
644+
chunkRange,
645+
timeDuplicateInfo,
646+
allValueColDeletedMap,
647+
maxNumberOfPointsInPage,
648+
activeSchemaList);
619649
}
620650

621651
private void handleEncoding(
622652
BlockingQueue<Object> ioTaskQueue,
623653
List<List<Integer>> chunkRange,
624654
boolean[] timeDuplicateInfo,
625655
BitMap allValueColDeletedMap,
626-
int maxNumberOfPointsInPage) {
656+
int maxNumberOfPointsInPage,
657+
List<IMeasurementSchema> activeSchemaList) {
627658
AlignedTVList alignedWorkingListForFlush = (AlignedTVList) workingListForFlush;
628-
List<TSDataType> dataTypes = alignedWorkingListForFlush.getTsDataTypes();
629-
Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new Pair[dataTypes.size()];
659+
List<Integer> columnIndexList = buildColumnIndexList(activeSchemaList);
660+
Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new Pair[activeSchemaList.size()];
630661
for (List<Integer> pageRange : chunkRange) {
631662
AlignedChunkWriterImpl alignedChunkWriter =
632-
new AlignedChunkWriterImpl(schemaList, encryptParameter);
663+
new AlignedChunkWriterImpl(activeSchemaList, encryptParameter);
633664
for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum += 1) {
634-
for (int columnIndex = 0; columnIndex < dataTypes.size(); columnIndex++) {
665+
for (int columnIndex = 0; columnIndex < activeSchemaList.size(); columnIndex++) {
666+
int tvListColumnIndex = columnIndexList.get(columnIndex);
635667
// Pair of Time and Index
636668
if (Objects.nonNull(timeDuplicateInfo)
637669
&& lastValidPointIndexForTimeDupCheck[columnIndex] == null) {
638670
lastValidPointIndexForTimeDupCheck[columnIndex] = new Pair<>(Long.MIN_VALUE, null);
639671
}
640-
TSDataType tsDataType = dataTypes.get(columnIndex);
672+
TSDataType tsDataType = activeSchemaList.get(columnIndex).getType();
641673
for (int sortedRowIndex = pageRange.get(pageNum * 2);
642674
sortedRowIndex <= pageRange.get(pageNum * 2 + 1);
643675
sortedRowIndex++) {
@@ -650,8 +682,10 @@ private void handleEncoding(
650682
// skip time duplicated rows
651683
long time = alignedWorkingListForFlush.getTime(sortedRowIndex);
652684
if (Objects.nonNull(timeDuplicateInfo)) {
653-
if (!alignedWorkingListForFlush.isNullValue(
654-
alignedWorkingListForFlush.getValueIndex(sortedRowIndex), columnIndex)) {
685+
if (tvListColumnIndex >= 0
686+
&& !alignedWorkingListForFlush.isNullValue(
687+
alignedWorkingListForFlush.getValueIndex(sortedRowIndex),
688+
tvListColumnIndex)) {
655689
lastValidPointIndexForTimeDupCheck[columnIndex].left = time;
656690
lastValidPointIndexForTimeDupCheck[columnIndex].right =
657691
alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
@@ -671,21 +705,24 @@ private void handleEncoding(
671705
// write(T:3,V:null)
672706

673707
int originRowIndex;
674-
if (Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex])
708+
if (tvListColumnIndex >= 0
709+
&& Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex])
675710
&& (time == lastValidPointIndexForTimeDupCheck[columnIndex].left)) {
676711
originRowIndex = lastValidPointIndexForTimeDupCheck[columnIndex].right;
677712
} else {
678713
originRowIndex = alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
679714
}
680715

681-
boolean isNull = alignedWorkingListForFlush.isNullValue(originRowIndex, columnIndex);
716+
boolean isNull =
717+
tvListColumnIndex < 0
718+
|| alignedWorkingListForFlush.isNullValue(originRowIndex, tvListColumnIndex);
682719
switch (tsDataType) {
683720
case BOOLEAN:
684721
alignedChunkWriter.writeByColumn(
685722
time,
686723
!isNull
687724
&& alignedWorkingListForFlush.getBooleanByValueIndex(
688-
originRowIndex, columnIndex),
725+
originRowIndex, tvListColumnIndex),
689726
isNull);
690727
break;
691728
case INT32:
@@ -695,7 +732,7 @@ private void handleEncoding(
695732
isNull
696733
? 0
697734
: alignedWorkingListForFlush.getIntByValueIndex(
698-
originRowIndex, columnIndex),
735+
originRowIndex, tvListColumnIndex),
699736
isNull);
700737
break;
701738
case INT64:
@@ -705,7 +742,7 @@ private void handleEncoding(
705742
isNull
706743
? 0
707744
: alignedWorkingListForFlush.getLongByValueIndex(
708-
originRowIndex, columnIndex),
745+
originRowIndex, tvListColumnIndex),
709746
isNull);
710747
break;
711748
case FLOAT:
@@ -714,7 +751,7 @@ private void handleEncoding(
714751
isNull
715752
? 0
716753
: alignedWorkingListForFlush.getFloatByValueIndex(
717-
originRowIndex, columnIndex),
754+
originRowIndex, tvListColumnIndex),
718755
isNull);
719756
break;
720757
case DOUBLE:
@@ -723,7 +760,7 @@ private void handleEncoding(
723760
isNull
724761
? 0
725762
: alignedWorkingListForFlush.getDoubleByValueIndex(
726-
originRowIndex, columnIndex),
763+
originRowIndex, tvListColumnIndex),
727764
isNull);
728765
break;
729766
case TEXT:
@@ -735,7 +772,7 @@ private void handleEncoding(
735772
isNull
736773
? null
737774
: alignedWorkingListForFlush.getBinaryByValueIndex(
738-
originRowIndex, columnIndex),
775+
originRowIndex, tvListColumnIndex),
739776
isNull);
740777
break;
741778
default:
@@ -787,16 +824,21 @@ public void encode(BlockingQueue<Object> ioTaskQueue, BatchEncodeInfo encodeInfo
787824
return;
788825
}
789826

827+
List<IMeasurementSchema> activeSchemaList = getActiveSchemaList();
828+
if (activeSchemaList.isEmpty()) {
829+
return;
830+
}
831+
790832
AlignedChunkWriterImpl alignedChunkWriter =
791-
new AlignedChunkWriterImpl(schemaList, encryptParameter);
833+
new AlignedChunkWriterImpl(activeSchemaList, encryptParameter);
792834

793835
// create MergeSortAlignedTVListIterator.
794836
List<AlignedTVList> alignedTvLists = new ArrayList<>(sortedList);
795837
alignedTvLists.add((AlignedTVList) workingListForFlush);
796-
List<Integer> columnIndexList = buildColumnIndexList(schemaList);
838+
List<Integer> columnIndexList = buildColumnIndexList(activeSchemaList);
797839
MemPointIterator timeValuePairIterator =
798840
MemPointIteratorFactory.create(
799-
dataTypes,
841+
getDataTypes(activeSchemaList),
800842
columnIndexList,
801843
alignedTvLists,
802844
ignoreAllNullRows,
@@ -818,7 +860,7 @@ public void encode(BlockingQueue<Object> ioTaskQueue, BatchEncodeInfo encodeInfo
818860
} catch (InterruptedException e) {
819861
Thread.currentThread().interrupt();
820862
}
821-
alignedChunkWriter = new AlignedChunkWriterImpl(schemaList, encryptParameter);
863+
alignedChunkWriter = new AlignedChunkWriterImpl(activeSchemaList, encryptParameter);
822864
encodeInfo.reset();
823865
}
824866
}

0 commit comments

Comments
 (0)