You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by ji...@apache.org on 2025/05/20 11:04:43 UTC
(tsfile) branch add_last_reader created (now 3b389a7c)
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a change to branch add_last_reader
in repository https://gitbox.apache.org/repos/asf/tsfile.git
at 3b389a7c Add TsFileLastReader for retrieving last points in a TsFile
This branch includes the following new commits:
new 3b389a7c Add TsFileLastReader for retrieving last points in a TsFile
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(tsfile) 01/01: Add TsFileLastReader for retrieving last points in a TsFile
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch add_last_reader
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 3b389a7c9c811a9f1d1362f55efce6efbb75aa5f
Author: Tian Jiang <jt...@163.com>
AuthorDate: Tue May 20 19:04:05 2025 +0800
Add TsFileLastReader for retrieving last points in a TsFile
---
cpp/pom.xml | 2 +-
java/common/pom.xml | 2 +-
java/examples/pom.xml | 4 +-
java/pom.xml | 4 +-
java/tools/pom.xml | 6 +-
java/tsfile/pom.xml | 4 +-
.../apache/tsfile/read/TsFileSequenceReader.java | 132 +++++++++++++++
.../tsfile/read/reader/TsFileLastReader.java | 177 +++++++++++++++++++++
.../org/apache/tsfile/write/record/Tablet.java | 14 +-
.../tsfile/read/reader/TsFileLastReaderTest.java | 144 +++++++++++++++++
pom.xml | 6 +-
python/pom.xml | 2 +-
12 files changed, 478 insertions(+), 19 deletions(-)
diff --git a/cpp/pom.xml b/cpp/pom.xml
index d39c67c0..cc62aa8c 100644
--- a/cpp/pom.xml
+++ b/cpp/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-parent</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</parent>
<artifactId>tsfile-cpp</artifactId>
<packaging>pom</packaging>
diff --git a/java/common/pom.xml b/java/common/pom.xml
index 0eb4066d..bcd54f5e 100644
--- a/java/common/pom.xml
+++ b/java/common/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-java</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</parent>
<artifactId>common</artifactId>
<name>TsFile: Java: Common</name>
diff --git a/java/examples/pom.xml b/java/examples/pom.xml
index 9223f596..5a484cfc 100644
--- a/java/examples/pom.xml
+++ b/java/examples/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-java</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</parent>
<artifactId>examples</artifactId>
<name>TsFile: Java: Examples</name>
@@ -36,7 +36,7 @@
<dependency>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</dependency>
</dependencies>
<build>
diff --git a/java/pom.xml b/java/pom.xml
index 1d99dbba..df7cec5a 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -24,10 +24,10 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-parent</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</parent>
<artifactId>tsfile-java</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
<packaging>pom</packaging>
<name>TsFile: Java</name>
<modules>
diff --git a/java/tools/pom.xml b/java/tools/pom.xml
index 0bc2c89a..8cc58d1a 100644
--- a/java/tools/pom.xml
+++ b/java/tools/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-java</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</parent>
<artifactId>tools</artifactId>
<name>TsFile: Java: Tools</name>
@@ -32,7 +32,7 @@
<dependency>
<groupId>org.apache.tsfile</groupId>
<artifactId>common</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>commons-cli</groupId>
@@ -50,7 +50,7 @@
<dependency>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
diff --git a/java/tsfile/pom.xml b/java/tsfile/pom.xml
index 5b22ea2e..8625f0cf 100644
--- a/java/tsfile/pom.xml
+++ b/java/tsfile/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-java</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</parent>
<artifactId>tsfile</artifactId>
<name>TsFile: Java: TsFile</name>
@@ -38,7 +38,7 @@
<dependency>
<groupId>org.apache.tsfile</groupId>
<artifactId>common</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.github.luben</groupId>
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/TsFileSequenceReader.java b/java/tsfile/src/main/java/org/apache/tsfile/read/TsFileSequenceReader.java
index dc07bfc2..1bf69c42 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/read/TsFileSequenceReader.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/read/TsFileSequenceReader.java
@@ -87,9 +87,11 @@ import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
+import java.util.Deque;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
@@ -1492,6 +1494,11 @@ public class TsFileSequenceReader implements AutoCloseable {
return timeseriesMetadataMap;
}
+ public Iterator<Pair<IDeviceID, List<TimeseriesMetadata>>> iterAllTimeseriesMetadata(
+ boolean needChunkMetadata) throws IOException {
+ return new TimeseriesMetadataIterator(needChunkMetadata);
+ }
+
/* This method will only deserialize the TimeseriesMetadata, not including chunk metadata list */
public List<TimeseriesMetadata> getDeviceTimeseriesMetadataWithoutChunkMetadata(IDeviceID device)
throws IOException {
@@ -2973,4 +2980,129 @@ public class TsFileSequenceReader implements AutoCloseable {
public DeserializeConfig getDeserializeContext() {
return deserializeConfig;
}
+
+ private class TimeseriesMetadataIterator
+ implements Iterator<Pair<IDeviceID, List<TimeseriesMetadata>>> {
+
+ private final Deque<MetadataIndexNode> nodeStack = new ArrayDeque<>();
+ private final boolean needChunkMetadata;
+ private Pair<IDeviceID, List<TimeseriesMetadata>> nextValue;
+ private MetadataIndexNode currentLeafDeviceNode;
+ private int currentLeafDeviceNodeIndex;
+
+ public TimeseriesMetadataIterator(boolean needChunkMetadata) throws IOException {
+ this.needChunkMetadata = needChunkMetadata;
+ if (tsFileMetaData == null) {
+ readFileMetadata();
+ }
+
+ nodeStack.addAll(tsFileMetaData.getTableMetadataIndexNodeMap().values());
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (nextValue != null) {
+ return true;
+ }
+
+ try {
+ loadNextValue();
+ } catch (IOException e) {
+ logger.warn("Cannot read timeseries metadata from {},", file, e);
+ return false;
+ }
+ return nextValue != null;
+ }
+
+ @Override
+ public Pair<IDeviceID, List<TimeseriesMetadata>> next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ Pair<IDeviceID, List<TimeseriesMetadata>> ret = nextValue;
+ nextValue = null;
+ return ret;
+ }
+
+ private void loadNextLeafDeviceNode() throws IOException {
+ while (!nodeStack.isEmpty()) {
+ MetadataIndexNode node = nodeStack.pop();
+ MetadataIndexNodeType nodeType = node.getNodeType();
+ if (nodeType.equals(MetadataIndexNodeType.LEAF_DEVICE)) {
+ currentLeafDeviceNode = node;
+ currentLeafDeviceNodeIndex = 0;
+ return;
+ }
+
+ List<IMetadataIndexEntry> childrenIndex = node.getChildren();
+ for (int i = 0; i < childrenIndex.size(); i++) {
+ long endOffset;
+ IMetadataIndexEntry childIndex = childrenIndex.get(i);
+ endOffset = node.getEndOffset();
+ if (i != childrenIndex.size() - 1) {
+ endOffset = childrenIndex.get(i + 1).getOffset();
+ }
+
+ MetadataIndexNode child;
+ if (endOffset - childIndex.getOffset() < Integer.MAX_VALUE) {
+ ByteBuffer buffer = readData(childIndex.getOffset(), endOffset);
+ child = deserializeConfig.deserializeMetadataIndexNode(buffer, true);
+ } else {
+ tsFileInput.position(childIndex.getOffset());
+ child =
+ deserializeConfig.deserializeMetadataIndexNode(
+ tsFileInput.wrapAsInputStream(), true);
+ }
+ nodeStack.push(child);
+ }
+ }
+ }
+
+ private void loadNextValue() throws IOException {
+ if (currentLeafDeviceNode == null
+ || currentLeafDeviceNodeIndex >= currentLeafDeviceNode.getChildren().size()) {
+ currentLeafDeviceNode = null;
+ loadNextLeafDeviceNode();
+ }
+ if (currentLeafDeviceNode == null) {
+ return;
+ }
+
+ IMetadataIndexEntry childIndex =
+ currentLeafDeviceNode.getChildren().get(currentLeafDeviceNodeIndex);
+ int childNum = currentLeafDeviceNode.getChildren().size();
+ IDeviceID deviceId = ((DeviceMetadataIndexEntry) childIndex).getDeviceID();
+
+ Map<IDeviceID, List<TimeseriesMetadata>> nextValueMap = new HashMap<>(1);
+ long endOffset = currentLeafDeviceNode.getEndOffset();
+ if (currentLeafDeviceNodeIndex != childNum - 1) {
+ endOffset =
+ currentLeafDeviceNode.getChildren().get(currentLeafDeviceNodeIndex + 1).getOffset();
+ }
+ if (endOffset - childIndex.getOffset() < Integer.MAX_VALUE) {
+ ByteBuffer nextBuffer = readData(childIndex.getOffset(), endOffset);
+ generateMetadataIndex(
+ childIndex,
+ nextBuffer,
+ deviceId,
+ currentLeafDeviceNode.getNodeType(),
+ nextValueMap,
+ needChunkMetadata);
+ } else {
+ // when the buffer length is over than Integer.MAX_VALUE,
+ // using tsFileInput to get timeseriesMetadataList
+ generateMetadataIndexUsingTsFileInput(
+ childIndex,
+ childIndex.getOffset(),
+ endOffset,
+ deviceId,
+ currentLeafDeviceNode.getNodeType(),
+ nextValueMap,
+ needChunkMetadata);
+ }
+ currentLeafDeviceNodeIndex++;
+ Entry<IDeviceID, List<TimeseriesMetadata>> entry = nextValueMap.entrySet().iterator().next();
+ nextValue = new Pair<>(entry.getKey(), entry.getValue());
+ }
+ }
}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/TsFileLastReader.java b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/TsFileLastReader.java
new file mode 100644
index 00000000..780164fc
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/TsFileLastReader.java
@@ -0,0 +1,177 @@
+/*
+ * 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.tsfile.read.reader;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.TimeseriesMetadata;
+import org.apache.tsfile.read.TimeValuePair;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.utils.TsPrimitiveType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinTask;
+import java.util.stream.Collectors;
+
+/** Conveniently retrieve last points of all timeseries from a TsFile. */
+public class TsFileLastReader
+ implements AutoCloseable, Iterator<Pair<IDeviceID, List<Pair<String, TimeValuePair>>>> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(TsFileLastReader.class);
+
+ private final TsFileSequenceReader sequenceReader;
+ private boolean asyncIO = true;
+ private Iterator<Pair<IDeviceID, List<TimeseriesMetadata>>> timeseriesMetadataIter;
+ private Pair<IDeviceID, List<Pair<String, TimeValuePair>>> nextValue;
+
+ private BlockingQueue<Pair<IDeviceID, List<Pair<String, TimeValuePair>>>> lastValueQueue;
+ private ForkJoinTask<Void> asyncTask;
+
+ public TsFileLastReader(String filePath) throws IOException {
+ sequenceReader = new TsFileSequenceReader(filePath);
+ }
+
+ public TsFileLastReader(String filePath, boolean asyncIO) throws IOException {
+ this(filePath);
+ this.asyncIO = asyncIO;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (timeseriesMetadataIter == null) {
+ try {
+ init();
+ } catch (IOException e) {
+ LOGGER.error("Cannot read timeseries metadata from {}", sequenceReader.getFileName(), e);
+ return false;
+ }
+ }
+
+ // already meet the terminator
+ if (nextValue != null) {
+ return nextValue.getLeft() != null;
+ }
+
+ if (asyncIO) {
+ try {
+ nextValue = lastValueQueue.take();
+ if (nextValue.getLeft() == null) {
+ // the terminator
+ return false;
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ } else {
+ if (!timeseriesMetadataIter.hasNext()) {
+ nextValue = new Pair<>(null, null);
+ } else {
+ Pair<IDeviceID, List<TimeseriesMetadata>> next = timeseriesMetadataIter.next();
+ nextValue = new Pair<>(next.left, convertToLastPoints(next.right));
+ }
+ }
+ return nextValue.left != null;
+ }
+
+ /**
+ * @return (deviceId, measurementId, lastPoint)
+ */
+ @Override
+ public Pair<IDeviceID, List<Pair<String, TimeValuePair>>> next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ Pair<IDeviceID, List<Pair<String, TimeValuePair>>> ret = nextValue;
+ nextValue = null;
+ return ret;
+ }
+
+ private List<Pair<String, TimeValuePair>> convertToLastPoints(
+ List<TimeseriesMetadata> timeseriesMetadataList) {
+ return timeseriesMetadataList.stream()
+ .map(
+ seriesMeta ->
+ new Pair<>(
+ seriesMeta.getMeasurementId(),
+ new TimeValuePair(
+ seriesMeta.getStatistics().getEndTime(),
+ TsPrimitiveType.getByType(
+ seriesMeta.getTsDataType() == TSDataType.VECTOR
+ ? TSDataType.INT64
+ : seriesMeta.getTsDataType(),
+ seriesMeta.getTsDataType() == TSDataType.VECTOR
+ ? seriesMeta.getStatistics().getEndTime()
+ : seriesMeta.getStatistics().getLastValue()))))
+ .collect(Collectors.toList());
+ }
+
+ private void init() throws IOException {
+ timeseriesMetadataIter = sequenceReader.iterAllTimeseriesMetadata(false);
+ if (asyncIO) {
+ int queueCapacity = 1024;
+ lastValueQueue = new ArrayBlockingQueue<>(queueCapacity);
+ asyncTask =
+ ForkJoinPool.commonPool()
+ .submit(
+ () -> {
+ try {
+ while (timeseriesMetadataIter.hasNext()) {
+ Pair<IDeviceID, List<TimeseriesMetadata>> deviceSeriesMetadata =
+ timeseriesMetadataIter.next();
+ lastValueQueue.put(
+ new Pair<>(
+ deviceSeriesMetadata.left,
+ convertToLastPoints(deviceSeriesMetadata.right)));
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ LOGGER.error("Error while reading timeseries metadata", e);
+ } finally {
+ try {
+ lastValueQueue.put(new Pair<>(null, null));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return null;
+ });
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (asyncIO) {
+ asyncTask.cancel(true);
+ }
+ sequenceReader.close();
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/record/Tablet.java b/java/tsfile/src/main/java/org/apache/tsfile/write/record/Tablet.java
index f64dc059..d84cd057 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/record/Tablet.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/record/Tablet.java
@@ -412,12 +412,18 @@ public class Tablet {
@TsFileApi
public void addValue(int rowIndex, int columnIndex, int val) {
- if (!(values[columnIndex] instanceof int[])) {
+ if (!(values[columnIndex] instanceof int[]) && !(values[columnIndex] instanceof long[])) {
throw new IllegalArgumentException(
- "The data type of column index " + columnIndex + " is not INT32");
+ "The data type of column index " + columnIndex + " is not INT32 or INT64");
}
- final int[] sensor = (int[]) values[columnIndex];
- sensor[rowIndex] = val;
+ if (values[columnIndex] instanceof int[]) {
+ final int[] sensor = (int[]) values[columnIndex];
+ sensor[rowIndex] = val;
+ } else if (values[columnIndex] instanceof long[]) {
+ final long[] sensor = (long[]) values[columnIndex];
+ sensor[rowIndex] = val;
+ }
+
updateBitMap(rowIndex, columnIndex, false);
}
diff --git a/java/tsfile/src/test/java/org/apache/tsfile/read/reader/TsFileLastReaderTest.java b/java/tsfile/src/test/java/org/apache/tsfile/read/reader/TsFileLastReaderTest.java
new file mode 100644
index 00000000..f3aad0c8
--- /dev/null
+++ b/java/tsfile/src/test/java/org/apache/tsfile/read/reader/TsFileLastReaderTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.tsfile.read.reader;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.write.WriteProcessException;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.TimeValuePair;
+import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.write.TsFileWriter;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+public class TsFileLastReaderTest {
+ private final String filePath = "target/test.tsfile";
+ private final File file = new File(filePath);
+
+ private void createFile(int deviceNum, int measurementNum, int seriesPointNum)
+ throws IOException, WriteProcessException {
+ try (TsFileWriter writer = new TsFileWriter(file)) {
+ List<IMeasurementSchema> measurementSchemaList = new ArrayList<>();
+ for (int j = 0; j < measurementNum; j++) {
+ measurementSchemaList.add(new MeasurementSchema("s" + j, TSDataType.INT64));
+ }
+ for (int i = 0; i < deviceNum; i++) {
+ writer.registerAlignedTimeseries("device" + i, measurementSchemaList);
+ }
+
+ for (int i = 0; i < deviceNum; i++) {
+ Tablet tablet = new Tablet("device" + i, measurementSchemaList, seriesPointNum);
+ for (int k = 0; k < seriesPointNum; k++) {
+ tablet.addTimestamp(k, k);
+ }
+ for (int j = 0; j < measurementNum; j++) {
+ for (int k = 0; k < seriesPointNum; k++) {
+ tablet.addValue(k, j, k);
+ }
+ }
+ writer.writeTree(tablet);
+ }
+ }
+ }
+
+ private void doReadLast(int deviceNum, int measurementNum, int seriesPointNum) throws Exception {
+ long startTime = System.currentTimeMillis();
+ Set<IDeviceID> devices = new HashSet<>();
+ try (TsFileLastReader lastReader = new TsFileLastReader(filePath, false)) {
+ while (lastReader.hasNext()) {
+ Set<String> measurements = new HashSet<>();
+ Pair<IDeviceID, List<Pair<String, TimeValuePair>>> next = lastReader.next();
+ assertFalse(devices.contains(next.left));
+ devices.add(next.left);
+
+ // time column included
+ assertEquals(measurementNum + 1, next.getRight().size());
+ next.right.forEach(
+ pair -> {
+ measurements.add(pair.getLeft());
+ assertEquals(seriesPointNum - 1, pair.getRight().getTimestamp());
+ assertEquals(seriesPointNum - 1, pair.getRight().getValue().getLong());
+ });
+ assertEquals(measurementNum + 1, measurements.size());
+ }
+ }
+ assertEquals(deviceNum, devices.size());
+ System.out.printf("Last point iteration takes %dms%n", System.currentTimeMillis() - startTime);
+ }
+
+ private void testReadLast(int deviceNum, int measurementNum, int seriesPointNum)
+ throws Exception {
+ createFile(deviceNum, measurementNum, seriesPointNum);
+ doReadLast(deviceNum, measurementNum, seriesPointNum);
+ file.delete();
+ }
+
+ @Test
+ public void testSmall() throws Exception {
+ testReadLast(10, 10, 10);
+ }
+
+ @Test
+ public void testManyDevices() throws Exception {
+ testReadLast(10000, 10, 10);
+ }
+
+ @Test
+ public void testManyMeasurement() throws Exception {
+ testReadLast(10, 10000, 10);
+ }
+
+ @Test
+ public void testManyPoints() throws Exception {
+ testReadLast(100, 10, 10000);
+ }
+
+ @Test
+ public void testManyMany() throws Exception {
+ testReadLast(1000, 1000, 1000);
+ }
+
+ @Ignore("Performance")
+ @Test
+ public void testManyRead() throws Exception {
+ int deviceNum = 10000;
+ int measurementNum = 1000;
+ int seriesPointNum = 1;
+ createFile(deviceNum, measurementNum, seriesPointNum);
+ for (int i = 0; i < 10; i++) {
+ doReadLast(deviceNum, measurementNum, seriesPointNum);
+ }
+ file.delete();
+ }
+}
diff --git a/pom.xml b/pom.xml
index 2e1d5533..08fb3655 100644
--- a/pom.xml
+++ b/pom.xml
@@ -28,7 +28,7 @@
</parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-parent</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Apache TsFile Project Parent POM</name>
<properties>
@@ -68,7 +68,7 @@
<artifactId>maven-surefire-plugin</artifactId>
<version>3.5.0</version>
<configuration>
- <argLine>${argLine} -Xmx1024m</argLine>
+ <argLine>${argLine}</argLine>
</configuration>
</plugin>
<!--
@@ -1021,7 +1021,7 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
- <argLine>${argLine} @{surefire.jacoco.args} -Xmx1024m</argLine>
+ <argLine>${argLine} @{surefire.jacoco.args}</argLine>
</configuration>
</plugin>
<!-- for IT-->
diff --git a/python/pom.xml b/python/pom.xml
index 051e77f0..88e48818 100644
--- a/python/pom.xml
+++ b/python/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-parent</artifactId>
- <version>2.1.0-250325-SNAPSHOT</version>
+ <version>2.1.0-SNAPSHOT</version>
</parent>
<artifactId>tsfile-python</artifactId>
<packaging>pom</packaging>