You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2024/12/25 13:46:44 UTC
(iotdb) branch meituan_rc/1.3.3 updated: Add the used memory calculation for DescPriorityMergeReader which is missed before (#14548)
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch meituan_rc/1.3.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/meituan_rc/1.3.3 by this push:
new 359af3e99ac Add the used memory calculation for DescPriorityMergeReader which is missed before (#14548)
359af3e99ac is described below
commit 359af3e99ac5b01eb2fe831f3a295b064d0402f8
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Dec 25 21:43:26 2024 +0800
Add the used memory calculation for DescPriorityMergeReader which is missed before (#14548)
---
.../reader/common/CachedPriorityMergeReader.java | 97 ------------------
.../reader/common/DescPriorityMergeReader.java | 16 +--
.../read/reader/common/PriorityMergeReader.java | 22 ++--
...SeriesReader.java => AscFakedSeriesReader.java} | 28 ++---
.../read/reader/common/DescFakedSeriesReader.java | 54 ++++++++++
.../reader/common/PriorityMergeReaderTest.java | 113 ++++++++++++++++++---
.../reader/common/PriorityMergeReaderTest2.java | 59 -----------
7 files changed, 177 insertions(+), 212 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/CachedPriorityMergeReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/CachedPriorityMergeReader.java
deleted file mode 100644
index 6b1d8cf4459..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/CachedPriorityMergeReader.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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.read.reader.common;
-
-import org.apache.iotdb.db.utils.TimeValuePairUtils;
-
-import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.read.TimeValuePair;
-
-import java.io.IOException;
-
-/**
- * CachedPriorityMergeReader use a cache to reduce unnecessary heap updates and increase locality.
- */
-public class CachedPriorityMergeReader extends PriorityMergeReader {
-
- private static final int CACHE_SIZE = 100;
-
- private TimeValuePair[] timeValuePairCache = new TimeValuePair[CACHE_SIZE];
- private int cacheLimit = 0;
- private int cacheIdx = 0;
-
- private Long lastTimestamp = null;
-
- public CachedPriorityMergeReader(TSDataType dataType) {
- for (int i = 0; i < CACHE_SIZE; i++) {
- timeValuePairCache[i] = TimeValuePairUtils.getEmptyTimeValuePair(dataType);
- }
- }
-
- @Override
- public boolean hasNextTimeValuePair() {
- return cacheIdx < cacheLimit || !heap.isEmpty();
- }
-
- private void fetch() throws IOException {
- cacheLimit = 0;
- cacheIdx = 0;
- while (!heap.isEmpty() && cacheLimit < CACHE_SIZE) {
- Element top = heap.peek();
- if (lastTimestamp == null || top.currTime() != lastTimestamp) {
- TimeValuePairUtils.setTimeValuePair(
- top.getTimeValuePair(), timeValuePairCache[cacheLimit++]);
- lastTimestamp = top.currTime();
- }
- // remove duplicates
- while (heap.peek() != null && heap.peek().currTime() == lastTimestamp) {
- top = heap.poll();
- if (top.hasNext()) {
- top.next();
- heap.add(top);
- } else {
- top.close();
- }
- }
- }
- }
-
- @Override
- public TimeValuePair nextTimeValuePair() throws IOException {
- TimeValuePair ret;
- if (cacheIdx < cacheLimit) {
- ret = timeValuePairCache[cacheIdx++];
- } else {
- fetch();
- ret = timeValuePairCache[cacheIdx++];
- }
- return ret;
- }
-
- @Override
- public TimeValuePair currentTimeValuePair() throws IOException {
- if (0 <= cacheIdx && cacheIdx < cacheLimit) {
- return timeValuePairCache[cacheIdx];
- } else {
- fetch();
- return timeValuePairCache[cacheIdx];
- }
- }
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/DescPriorityMergeReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/DescPriorityMergeReader.java
index c6975b77a98..2f411bc6f6a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/DescPriorityMergeReader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/DescPriorityMergeReader.java
@@ -19,15 +19,13 @@
package org.apache.iotdb.db.storageengine.dataregion.read.reader.common;
-import org.apache.tsfile.read.reader.IPointReader;
-
-import java.io.IOException;
import java.util.PriorityQueue;
public class DescPriorityMergeReader extends PriorityMergeReader {
public DescPriorityMergeReader() {
- super.heap =
+ currentReadStopTime = Long.MAX_VALUE;
+ heap =
new PriorityQueue<>(
(o1, o2) -> {
int timeCompare =
@@ -37,13 +35,7 @@ public class DescPriorityMergeReader extends PriorityMergeReader {
}
@Override
- public void addReader(IPointReader reader, MergeReaderPriority priority, long endTime)
- throws IOException {
- if (reader.hasNextTimeValuePair()) {
- heap.add(new Element(reader, reader.nextTimeValuePair(), priority));
- super.currentReadStopTime = Math.min(currentReadStopTime, endTime);
- } else {
- reader.close();
- }
+ protected void updateCurrentReadStopTime(long endTime) {
+ currentReadStopTime = Math.min(currentReadStopTime, endTime);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReader.java
index 7bcaab90e43..2f9cda7f30f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReader.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.storageengine.dataregion.read.reader.common;
-import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.queryengine.plan.planner.memory.MemoryReservationManager;
import org.apache.tsfile.read.TimeValuePair;
@@ -34,7 +33,7 @@ public class PriorityMergeReader implements IPointReader {
// max time of all added readers in PriorityMergeReader
// or min time of all added readers in DescPriorityMergeReader
- protected long currentReadStopTime;
+ protected long currentReadStopTime = Long.MIN_VALUE;
protected PriorityQueue<Element> heap;
@@ -57,25 +56,12 @@ public class PriorityMergeReader implements IPointReader {
this.memoryReservationManager = memoryReservationManager;
}
- @TestOnly
- public void addReader(IPointReader reader, long priority) throws IOException {
- if (reader.hasNextTimeValuePair()) {
- heap.add(
- new Element(
- reader,
- reader.nextTimeValuePair(),
- new MergeReaderPriority(Long.MAX_VALUE, priority, 0, false)));
- } else {
- reader.close();
- }
- }
-
public void addReader(IPointReader reader, MergeReaderPriority priority, long endTime)
throws IOException {
if (reader.hasNextTimeValuePair()) {
Element element = new Element(reader, reader.nextTimeValuePair(), priority);
heap.add(element);
- currentReadStopTime = Math.max(currentReadStopTime, endTime);
+ updateCurrentReadStopTime(endTime);
long size = element.getReader().getUsedMemorySize();
usedMemorySize += size;
if (memoryReservationManager != null) {
@@ -86,6 +72,10 @@ public class PriorityMergeReader implements IPointReader {
}
}
+ protected void updateCurrentReadStopTime(long endTime) {
+ currentReadStopTime = Math.max(currentReadStopTime, endTime);
+ }
+
public long getCurrentReadStopTime() {
return currentReadStopTime;
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FakedSeriesReader.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AscFakedSeriesReader.java
similarity index 79%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FakedSeriesReader.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AscFakedSeriesReader.java
index 0a183b62fe4..574b2544d66 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FakedSeriesReader.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AscFakedSeriesReader.java
@@ -25,23 +25,23 @@ import org.apache.tsfile.utils.TsPrimitiveType;
import java.io.IOException;
-public class FakedSeriesReader implements IPointReader {
+public class AscFakedSeriesReader implements IPointReader {
- private int index;
- private int size;
- private boolean initWithTimeList;
- private static final TSDataType DATA_TYPE = TSDataType.INT64;
+ protected int index;
+ protected int size;
+ protected boolean initWithTimeList;
+ protected static final TSDataType DATA_TYPE = TSDataType.INT64;
// init with time list and value
- private long[] timestamps;
- private long value;
+ protected long[] timestamps;
+ protected long value;
// init with startTime, size, interval and modValue
- private long startTime;
- private int interval;
- private int modValue;
+ protected long startTime;
+ protected int interval;
+ protected int modValue;
- public FakedSeriesReader(long[] timestamps, long value) {
+ public AscFakedSeriesReader(long[] timestamps, long value) {
this.initWithTimeList = true;
this.index = 0;
this.size = timestamps.length;
@@ -49,7 +49,7 @@ public class FakedSeriesReader implements IPointReader {
this.value = value;
}
- public FakedSeriesReader(long startTime, int size, int interval, int modValue) {
+ public AscFakedSeriesReader(long startTime, int size, int interval, int modValue) {
this.initWithTimeList = false;
this.index = 0;
this.size = size;
@@ -82,8 +82,8 @@ public class FakedSeriesReader implements IPointReader {
@Override
public long getUsedMemorySize() {
- // not used
- return 0;
+ // use size of timestamps to mock the used memory
+ return size;
}
@Override
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/DescFakedSeriesReader.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/DescFakedSeriesReader.java
new file mode 100644
index 00000000000..2135bbe5753
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/DescFakedSeriesReader.java
@@ -0,0 +1,54 @@
+/*
+ * 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.read.reader.common;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.TimeValuePair;
+import org.apache.tsfile.utils.TsPrimitiveType;
+
+public class DescFakedSeriesReader extends AscFakedSeriesReader {
+
+ public DescFakedSeriesReader(long[] timestamps, long value) {
+ super(timestamps, value);
+ index = size - 1;
+ }
+
+ public DescFakedSeriesReader(long startTime, int size, int interval, int modValue) {
+ super(startTime, size, interval, modValue);
+ index = size - 1;
+ startTime = startTime + interval * size;
+ }
+
+ @Override
+ public boolean hasNextTimeValuePair() {
+ return index >= 0;
+ }
+
+ @Override
+ public TimeValuePair nextTimeValuePair() {
+ if (initWithTimeList) {
+ return new TimeValuePair(timestamps[index--], TsPrimitiveType.getByType(DATA_TYPE, value));
+ } else {
+ long time = startTime;
+ startTime -= interval;
+ index--;
+ return new TimeValuePair(time, TsPrimitiveType.getByType(TSDataType.INT64, time % modValue));
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReaderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReaderTest.java
index 0c4b782aa09..fb6e5c88fc7 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReaderTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReaderTest.java
@@ -20,46 +20,46 @@
package org.apache.iotdb.db.storageengine.dataregion.read.reader.common;
import org.apache.tsfile.read.TimeValuePair;
-import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
+import static org.junit.Assert.assertEquals;
+
public class PriorityMergeReaderTest {
@Test
- public void test2S() throws IOException {
-
+ public void test1() throws IOException {
// 2 series
- test(
+ testAscAndDescPriorityMergeReader(
new long[] {1, 2, 3, 4, 5, 6},
new long[] {2, 2, 2, 1, 2, 2},
new long[] {1, 2, 3, 4, 5},
new long[] {1, 2, 3, 5, 6});
- test(
+ testAscAndDescPriorityMergeReader(
new long[] {1, 2, 3, 4, 5},
new long[] {1, 1, 1, 1, 1},
new long[] {1, 2, 3, 4, 5},
new long[] {});
- test(
+ testAscAndDescPriorityMergeReader(
new long[] {1, 2, 3, 4, 5},
new long[] {2, 2, 2, 2, 2},
new long[] {},
new long[] {1, 2, 3, 4, 5});
- test(
+ testAscAndDescPriorityMergeReader(
new long[] {1, 2, 3, 4, 5, 6, 7, 8},
new long[] {1, 1, 1, 1, 1, 2, 2, 2},
new long[] {1, 2, 3, 4, 5},
new long[] {6, 7, 8});
// 3 series
- test(
+ testAscAndDescPriorityMergeReader(
new long[] {1, 2, 3, 4, 5, 6, 7},
new long[] {3, 3, 3, 1, 3, 2, 3},
new long[] {1, 2, 3, 4, 5},
new long[] {1, 2, 3, 5, 6},
new long[] {1, 2, 3, 5, 7});
- test(
+ testAscAndDescPriorityMergeReader(
new long[] {1, 2, 3, 4, 5, 6},
new long[] {1, 1, 2, 3, 2, 3},
new long[] {1, 2},
@@ -67,18 +67,103 @@ public class PriorityMergeReaderTest {
new long[] {4, 6});
}
- private void test(long[] retTimestamp, long[] retValue, long[]... sources) throws IOException {
+ private void testAscAndDescPriorityMergeReader(
+ long[] retTimestamp, long[] retValue, long[]... sourceTimestamps) throws IOException {
+ testAsc(retTimestamp, retValue, sourceTimestamps);
+ testDesc(retTimestamp, retValue, sourceTimestamps);
+ }
+
+ private void testAsc(long[] retTimestamp, long[] retValue, long[]... sourceTimestamps)
+ throws IOException {
PriorityMergeReader priorityMergeReader = new PriorityMergeReader();
- for (int i = 0; i < sources.length; i++) {
- priorityMergeReader.addReader(new FakedSeriesReader(sources[i], i + 1), i + 1);
+ long mockUsedMemory = 0;
+ for (int i = 0; i < sourceTimestamps.length; i++) {
+ // priorityMergeReader.addReader(new FakedSeriesReader(sourceTimestamps[i], i + 1), i + 1);
+ long endTime =
+ sourceTimestamps[i].length == 0
+ ? Long.MIN_VALUE
+ : sourceTimestamps[i][sourceTimestamps[i].length - 1];
+ priorityMergeReader.addReader(
+ new AscFakedSeriesReader(sourceTimestamps[i], i + 1),
+ new MergeReaderPriority(Long.MAX_VALUE, i + 1, 0, false),
+ endTime);
+ mockUsedMemory += sourceTimestamps[i].length;
}
+ // use the size of all timestamps to mock the used memory
+ assertEquals(mockUsedMemory, priorityMergeReader.getUsedMemorySize());
+
int i = 0;
while (priorityMergeReader.hasNextTimeValuePair()) {
TimeValuePair timeValuePair = priorityMergeReader.nextTimeValuePair();
- Assert.assertEquals(retTimestamp[i], timeValuePair.getTimestamp());
- Assert.assertEquals(retValue[i], timeValuePair.getValue().getValue());
+ assertEquals(retTimestamp[i], timeValuePair.getTimestamp());
+ assertEquals(retValue[i], timeValuePair.getValue().getValue());
i++;
}
}
+
+ private void testDesc(long[] retTimestamp, long[] retValue, long[]... sourceTimestamps)
+ throws IOException {
+ PriorityMergeReader descPriorityMergeReader = new DescPriorityMergeReader();
+ long mockUsedMemory = 0;
+ for (int i = 0; i < sourceTimestamps.length; i++) {
+ // priorityMergeReader.addReader(new FakedSeriesReader(sourceTimestamps[i], i + 1), i + 1);
+ long endTime =
+ sourceTimestamps[i].length == 0
+ ? Long.MIN_VALUE
+ : sourceTimestamps[i][sourceTimestamps[i].length - 1];
+ descPriorityMergeReader.addReader(
+ new DescFakedSeriesReader(sourceTimestamps[i], i + 1),
+ new MergeReaderPriority(Long.MAX_VALUE, i + 1, 0, false),
+ endTime);
+ mockUsedMemory += sourceTimestamps[i].length;
+ }
+
+ // use the size of all timestamps to mock the used memory
+ assertEquals(mockUsedMemory, descPriorityMergeReader.getUsedMemorySize());
+
+ int i = retTimestamp.length - 1;
+ while (descPriorityMergeReader.hasNextTimeValuePair()) {
+ TimeValuePair timeValuePair = descPriorityMergeReader.nextTimeValuePair();
+ assertEquals(retTimestamp[i], timeValuePair.getTimestamp());
+ assertEquals(retValue[i], timeValuePair.getValue().getValue());
+ i--;
+ }
+ }
+
+ @Test
+ public void testAscPriorityMergeReader2() throws IOException {
+ AscFakedSeriesReader reader1 = new AscFakedSeriesReader(100, 80, 5, 11);
+ AscFakedSeriesReader reader2 = new AscFakedSeriesReader(150, 60, 6, 19);
+ AscFakedSeriesReader reader3 = new AscFakedSeriesReader(180, 50, 7, 31);
+
+ PriorityMergeReader priorityMergeReader = new PriorityMergeReader();
+ priorityMergeReader.addReader(
+ reader1, new MergeReaderPriority(Long.MAX_VALUE, 3, 0, false), 500);
+ priorityMergeReader.addReader(
+ reader2, new MergeReaderPriority(Long.MAX_VALUE, 2, 0, false), 510);
+ priorityMergeReader.addReader(
+ reader3, new MergeReaderPriority(Long.MAX_VALUE, 1, 0, false), 530);
+
+ // use the size of all timestamps to mock the used memory
+ assertEquals(80 + 60 + 50, priorityMergeReader.getUsedMemorySize());
+
+ int cnt = 0;
+ while (priorityMergeReader.hasNextTimeValuePair()) {
+ TimeValuePair timeValuePair = priorityMergeReader.nextTimeValuePair();
+ long time = timeValuePair.getTimestamp();
+ long value = (long) timeValuePair.getValue().getValue();
+
+ // System.out.println(time + "," + value);
+ if (time <= 500 && (time - 100) % 5 == 0) {
+ assertEquals(time % 11, value);
+ } else if (time <= 510 && (time - 150) % 6 == 0) {
+ assertEquals(time % 19, value);
+ } else {
+ assertEquals(time % 31, value);
+ }
+ cnt++;
+ }
+ assertEquals(162, cnt);
+ }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReaderTest2.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReaderTest2.java
deleted file mode 100644
index 8685e9d8f95..00000000000
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReaderTest2.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * 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.read.reader.common;
-
-import org.apache.tsfile.read.TimeValuePair;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-
-public class PriorityMergeReaderTest2 {
-
- @Test
- public void test() throws IOException {
- FakedSeriesReader reader1 = new FakedSeriesReader(100, 80, 5, 11);
- FakedSeriesReader reader2 = new FakedSeriesReader(150, 60, 6, 19);
- FakedSeriesReader reader3 = new FakedSeriesReader(180, 50, 7, 31);
-
- PriorityMergeReader priorityMergeReader = new PriorityMergeReader();
- priorityMergeReader.addReader(reader1, 3);
- priorityMergeReader.addReader(reader2, 2);
- priorityMergeReader.addReader(reader3, 1);
-
- int cnt = 0;
- while (priorityMergeReader.hasNextTimeValuePair()) {
- TimeValuePair timeValuePair = priorityMergeReader.nextTimeValuePair();
- long time = timeValuePair.getTimestamp();
- long value = (long) timeValuePair.getValue().getValue();
-
- // System.out.println(time + "," + value);
- if (time <= 500 && (time - 100) % 5 == 0) {
- Assert.assertEquals(time % 11, value);
- } else if (time <= 510 && (time - 150) % 6 == 0) {
- Assert.assertEquals(time % 19, value);
- } else {
- Assert.assertEquals(time % 31, value);
- }
- cnt++;
- }
- Assert.assertEquals(162, cnt);
- }
-}