This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch rel/1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.2 by this push:
new b85f7c76318 [To rel/1.2] Fix point priority reader index out of bound
(#10932)
b85f7c76318 is described below
commit b85f7c76318fa1c58db7a859ccb63bef68c94ab1
Author: shuwenwei <[email protected]>
AuthorDate: Tue Aug 29 11:49:46 2023 +0800
[To rel/1.2] Fix point priority reader index out of bound (#10932)
---
.../executor/fast/SeriesCompactionExecutor.java | 11 +-
.../utils/executor/fast/element/PointElement.java | 3 +
.../execute/utils/reader/PointPriorityReader.java | 14 ++-
.../FastCompactionPerformerWithEmptyPageTest.java | 137 +++++++++++++++++++++
.../compaction/utils/CompactionTestFileWriter.java | 99 ++++++++++++++-
5 files changed, 257 insertions(+), 7 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/SeriesCompactionExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/SeriesCompactionExecutor.java
index 33a401c67ef..a879a85ed66 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/SeriesCompactionExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/SeriesCompactionExecutor.java
@@ -244,8 +244,9 @@ public abstract class SeriesCompactionExecutor {
|| firstPageElement.needForceDecoding) {
// has overlap or modified pages, then deserialize it
summary.pageOverlapOrModified += 1;
- pointPriorityReader.addNewPage(firstPageElement);
- compactWithOverlapPages();
+ if (pointPriorityReader.addNewPageIfPageNotEmpty(firstPageElement)) {
+ compactWithOverlapPages();
+ }
} else {
// has none overlap or modified pages, flush it to chunk writer
directly
summary.pageNoneOverlap += 1;
@@ -276,7 +277,9 @@ public abstract class SeriesCompactionExecutor {
} else {
// unsealed page is not large enough or page.endTime > file.endTime,
then deserialze it
summary.pageNoneOverlapButDeserialize += 1;
- pointPriorityReader.addNewPage(pageElement);
+ if (!pointPriorityReader.addNewPageIfPageNotEmpty(pageElement)) {
+ return;
+ }
// write data points of the current page into chunk writer
TimeValuePair point;
@@ -352,7 +355,7 @@ public abstract class SeriesCompactionExecutor {
|| nextPageElement.needForceDecoding) {
// next page is overlapped or modified, then deserialize it
summary.pageOverlapOrModified++;
- pointPriorityReader.addNewPage(nextPageElement);
+ pointPriorityReader.addNewPageIfPageNotEmpty(nextPageElement);
} else {
// has none overlap or modified pages, flush it to chunk writer
directly
summary.pageFakeOverlap += 1;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/element/PointElement.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/element/PointElement.java
index 94f9bb0d031..eadae73d012 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/element/PointElement.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/element/PointElement.java
@@ -41,6 +41,9 @@ public class PointElement {
} else {
this.pointReader = pageElement.batchData.getTsBlockAlignedRowIterator();
}
+ if (!pointReader.hasNextTimeValuePair()) {
+ return;
+ }
this.timeValuePair = pointReader.nextTimeValuePair();
this.timestamp = timeValuePair.getTimestamp();
this.priority = pageElement.priority;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/PointPriorityReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/PointPriorityReader.java
index 5b66d66d70a..392b5141f13 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/PointPriorityReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/PointPriorityReader.java
@@ -170,8 +170,10 @@ public class PointPriorityReader {
* Add a new overlapped page.
*
* @throws IOException if io errors occurred
+ * @return whether page is added into the queue
*/
- public void addNewPage(PageElement pageElement) throws IOException {
+ public boolean addNewPageIfPageNotEmpty(PageElement pageElement)
+ throws IOException, IllegalPathException, WriteProcessException {
if (currentPointElement != null) {
nextPointInOtherPage = Math.min(nextPointInOtherPage,
pageElement.startTime);
if (currentPoint.getTimestamp() >= nextPointInOtherPage) {
@@ -179,6 +181,14 @@ public class PointPriorityReader {
currentPointElement = null;
}
}
- pointQueue.add(new PointElement(pageElement));
+ PointElement pointElement = new PointElement(pageElement);
+ boolean pageIsNotEmpty = pointElement.timeValuePair != null;
+ if (pageIsNotEmpty) {
+ pointQueue.add(pointElement);
+ } else {
+ removePage.call(pageElement);
+ }
+
+ return pageIsNotEmpty;
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithEmptyPageTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithEmptyPageTest.java
new file mode 100644
index 00000000000..815105aff1a
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithEmptyPageTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.storageengine.dataregion.compaction;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CrossSpaceCompactionTask;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionTestFileWriter;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class FastCompactionPerformerWithEmptyPageTest extends
AbstractCompactionTest {
+
+ @Before
+ public void setUp()
+ throws IOException, WriteProcessException, MetadataException,
InterruptedException {
+ super.setUp();
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ }
+
+ @Test
+ public void test1() throws IOException, IllegalPathException {
+ String device = "root.testsg.d1";
+ TsFileResource seqFile1 = createEmptyFileAndResource(true);
+ try (CompactionTestFileWriter writer = new
CompactionTestFileWriter(seqFile1)) {
+ writer.startChunkGroup("d1");
+ writer.generateSimpleAlignedSeriesToCurrentDeviceWithNullValue(
+ Arrays.asList("s1", "s2", "s3"),
+ new TimeRange[][] {new TimeRange[] {new TimeRange(10, 30)}},
+ TSEncoding.RLE,
+ CompressionType.UNCOMPRESSED,
+ Arrays.asList(false, true, true));
+ writer.generateSimpleAlignedSeriesToCurrentDeviceWithNullValue(
+ Arrays.asList("s1", "s2", "s3"),
+ new TimeRange[][] {new TimeRange[] {new TimeRange(40, 50)}},
+ TSEncoding.RLE,
+ CompressionType.UNCOMPRESSED,
+ Arrays.asList(false, false, false));
+ writer.endChunkGroup();
+ writer.endFile();
+ }
+ seqFile1.updateStartTime(device, 10);
+ seqFile1.updateEndTime(device, 50);
+ seqFile1.serialize();
+ generateModsFile(Arrays.asList(new PartialPath("root.testsg.d1.s1")),
seqFile1, 0, 31);
+
+ TsFileResource unseqFile1 = createEmptyFileAndResource(false);
+ try (CompactionTestFileWriter writer = new
CompactionTestFileWriter(unseqFile1)) {
+ writer.startChunkGroup("d1");
+ writer.generateSimpleAlignedSeriesToCurrentDevice(
+ Arrays.asList("s1", "s2", "s3"),
+ new TimeRange[][] {new TimeRange[] {new TimeRange(20, 34)}},
+ TSEncoding.RLE,
+ CompressionType.UNCOMPRESSED);
+ writer.endChunkGroup();
+ writer.endFile();
+ }
+ unseqFile1.updateStartTime(device, 20);
+ unseqFile1.updateEndTime(device, 34);
+ unseqFile1.serialize();
+
+ CrossSpaceCompactionTask task =
+ new CrossSpaceCompactionTask(
+ 0,
+ tsFileManager,
+ Arrays.asList(seqFile1),
+ Arrays.asList(unseqFile1),
+ new FastCompactionPerformer(true),
+ new AtomicInteger(0),
+ 0,
+ 0);
+ try {
+ Assert.assertTrue(task.start());
+ } catch (Exception e) {
+ Assert.fail();
+ }
+ TsFileResource result = tsFileManager.getTsFileList(true).get(0);
+ result.buildDeviceTimeIndex();
+ Assert.assertEquals(20, result.getStartTime(device));
+ Assert.assertEquals(50, result.getEndTime(device));
+
+ validateSeqFiles(true);
+
+ try (TsFileSequenceReader reader = new
TsFileSequenceReader(result.getTsFilePath())) {
+ Map<String, List<ChunkMetadata>> chunkMetadataInDevice =
+ reader.readChunkMetadataInDevice(device);
+ long startTime = Long.MAX_VALUE, endTime = Long.MIN_VALUE;
+ List<ChunkMetadata> chunkMetadataList = chunkMetadataInDevice.get("s1");
+ for (ChunkMetadata chunkMetadata : chunkMetadataList) {
+ startTime = Math.min(startTime, chunkMetadata.getStartTime());
+ endTime = Math.max(endTime, chunkMetadata.getEndTime());
+ }
+ Assert.assertEquals(20, startTime);
+ Assert.assertEquals(50, endTime);
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriter.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriter.java
index 8e447892107..fd02862199b 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriter.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriter.java
@@ -31,12 +31,13 @@ import
org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
+import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
-public class CompactionTestFileWriter {
+public class CompactionTestFileWriter implements Closeable {
private TsFileResource resource;
private TsFileIOWriter fileWriter;
@@ -170,6 +171,34 @@ public class CompactionTestFileWriter {
}
}
+ public void generateSimpleAlignedSeriesToCurrentDeviceWithNullValue(
+ List<String> measurementNames,
+ TimeRange[] toGenerateChunkTimeRanges,
+ TSEncoding encoding,
+ CompressionType compressionType,
+ List<Boolean> nullMeasurements)
+ throws IOException {
+ List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
+ for (String measurementName : measurementNames) {
+ measurementSchemas.add(
+ new MeasurementSchema(measurementName, TSDataType.INT32, encoding,
compressionType));
+ }
+ for (TimeRange toGenerateChunk : toGenerateChunkTimeRanges) {
+ AlignedChunkWriterImpl alignedChunkWriter = new
AlignedChunkWriterImpl(measurementSchemas);
+ currentDeviceStartTime = Math.min(toGenerateChunk.getMin(),
currentDeviceStartTime);
+ currentDeviceEndTime = Math.max(toGenerateChunk.getMax(),
currentDeviceEndTime);
+ for (long time = toGenerateChunk.getMin(); time <=
toGenerateChunk.getMax(); time++) {
+ alignedChunkWriter.getTimeChunkWriter().write(time);
+ for (int i = 0; i < measurementNames.size(); i++) {
+ alignedChunkWriter
+ .getValueChunkWriterByIndex(i)
+ .write(time, new Random().nextInt(), nullMeasurements.get(i));
+ }
+ }
+ alignedChunkWriter.writeToFileWriter(fileWriter);
+ }
+ }
+
public void generateSimpleAlignedSeriesToCurrentDevice(
List<String> measurementNames,
TimeRange[][] toGenerateChunkPageTimeRanges,
@@ -203,6 +232,40 @@ public class CompactionTestFileWriter {
}
}
+ public void generateSimpleAlignedSeriesToCurrentDeviceWithNullValue(
+ List<String> measurementNames,
+ TimeRange[][] toGenerateChunkPageTimeRanges,
+ TSEncoding encoding,
+ CompressionType compressionType,
+ List<Boolean> nullMeasurement)
+ throws IOException {
+ List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
+ for (String measurementName : measurementNames) {
+ measurementSchemas.add(
+ new MeasurementSchema(measurementName, TSDataType.INT32, encoding,
compressionType));
+ }
+ for (TimeRange[] toGenerateChunk : toGenerateChunkPageTimeRanges) {
+ AlignedChunkWriterImpl alignedChunkWriter = new
AlignedChunkWriterImpl(measurementSchemas);
+ for (TimeRange toGeneratePageTimeRange : toGenerateChunk) {
+ currentDeviceStartTime = Math.min(toGeneratePageTimeRange.getMin(),
currentDeviceStartTime);
+ currentDeviceEndTime = Math.max(toGeneratePageTimeRange.getMax(),
currentDeviceEndTime);
+ for (long time = toGeneratePageTimeRange.getMin();
+ time <= toGeneratePageTimeRange.getMax();
+ time++) {
+ alignedChunkWriter.write(time);
+ for (int i = 0; i < measurementNames.size(); i++) {
+ alignedChunkWriter
+ .getValueChunkWriterByIndex(i)
+ .getPageWriter()
+ .write(time, new Random().nextInt(), nullMeasurement.get(i));
+ }
+ }
+ alignedChunkWriter.sealCurrentPage();
+ }
+ alignedChunkWriter.writeToFileWriter(fileWriter);
+ }
+ }
+
public void generateSimpleAlignedSeriesToCurrentDevice(
List<String> measurementNames,
TimeRange[][][] toGenerateChunkPageTimeRanges,
@@ -235,4 +298,38 @@ public class CompactionTestFileWriter {
alignedChunkWriter.writeToFileWriter(fileWriter);
}
}
+
+ public void generateSimpleAlignedSeriesToCurrentDeviceWithNullValue(
+ List<String> measurementNames,
+ TimeRange[][][] toGenerateChunkPageTimeRanges,
+ TSEncoding encoding,
+ CompressionType compressionType,
+ List<Boolean> nullMeasurements)
+ throws IOException {
+ List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
+ for (String measurementName : measurementNames) {
+ measurementSchemas.add(
+ new MeasurementSchema(measurementName, TSDataType.INT32, encoding,
compressionType));
+ }
+ for (TimeRange[][] toGenerateChunk : toGenerateChunkPageTimeRanges) {
+ AlignedChunkWriterImpl alignedChunkWriter = new
AlignedChunkWriterImpl(measurementSchemas);
+ for (TimeRange[] toGeneratePageTimeRanges : toGenerateChunk) {
+ for (TimeRange pointsTimeRange : toGeneratePageTimeRanges) {
+ currentDeviceStartTime = Math.min(pointsTimeRange.getMin(),
currentDeviceStartTime);
+ currentDeviceEndTime = Math.max(pointsTimeRange.getMax(),
currentDeviceEndTime);
+ for (long time = pointsTimeRange.getMin(); time <=
pointsTimeRange.getMax(); time++) {
+ alignedChunkWriter.write(time);
+ for (int i = 0; i < measurementNames.size(); i++) {
+ alignedChunkWriter
+ .getValueChunkWriterByIndex(i)
+ .getPageWriter()
+ .write(time, new Random().nextInt(),
nullMeasurements.get(i));
+ }
+ }
+ }
+ alignedChunkWriter.sealCurrentPage();
+ }
+ alignedChunkWriter.writeToFileWriter(fileWriter);
+ }
+ }
}
