You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by su...@apache.org on 2022/01/10 03:42:23 UTC
[iotdb] 01/01: [IOTDB-2322] Provide a new interface to get AlignedChunkMetadata of device
This is an automated email from the ASF dual-hosted git repository.
sunzesong pushed a commit to branch jira-2322
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3e0417354ce5c18ae6e5976e63c9525c242b6e6d
Author: Zesong Sun <v-...@microsoft.com>
AuthorDate: Mon Jan 10 11:41:41 2022 +0800
[IOTDB-2322] Provide a new interface to get AlignedChunkMetadata of device
---
.../iotdb/tsfile/read/TsFileSequenceReader.java | 87 +++++++++++++++++++++-
.../tsfile/read/GetAlignedChunkMetadataTest.java | 72 ++++++++++++++++++
2 files changed, 156 insertions(+), 3 deletions(-)
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
index 573fa6f..8666812 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.tsfile.file.MetaMarker;
import org.apache.iotdb.tsfile.file.header.ChunkGroupHeader;
import org.apache.iotdb.tsfile.file.header.ChunkHeader;
import org.apache.iotdb.tsfile.file.header.PageHeader;
+import org.apache.iotdb.tsfile.file.metadata.AlignedChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.AlignedTimeSeriesMetadata;
import org.apache.iotdb.tsfile.file.metadata.ChunkGroupMetadata;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
@@ -389,7 +390,7 @@ public class TsFileSequenceReader implements AutoCloseable {
}
ByteBuffer buffer = readData(metadataIndexPair.left.getOffset(), metadataIndexPair.right);
MetadataIndexNode metadataIndexNode;
- TimeseriesMetadata firstTimeseriesMetadata = null;
+ TimeseriesMetadata firstTimeseriesMetadata;
try {
// next layer MeasurementNode of the specific DeviceNode
metadataIndexNode = MetadataIndexNode.deserializeFrom(buffer);
@@ -733,7 +734,7 @@ public class TsFileSequenceReader implements AutoCloseable {
Queue<Pair<String, Pair<Long, Long>>> queue = new LinkedList<>();
for (int i = 0; i < metadataIndexEntryList.size(); i++) {
MetadataIndexEntry metadataIndexEntry = metadataIndexEntryList.get(i);
- long endOffset = tsFileMetaData.getMetadataIndex().getEndOffset();
+ long endOffset = metadataIndexNode.getEndOffset();
if (i != metadataIndexEntryList.size() - 1) {
endOffset = metadataIndexEntryList.get(i + 1).getOffset();
}
@@ -905,7 +906,7 @@ public class TsFileSequenceReader implements AutoCloseable {
List<MetadataIndexEntry> metadataIndexEntryList = metadataIndexNode.getChildren();
for (int i = 0; i < metadataIndexEntryList.size(); i++) {
MetadataIndexEntry metadataIndexEntry = metadataIndexEntryList.get(i);
- long endOffset = tsFileMetaData.getMetadataIndex().getEndOffset();
+ long endOffset = metadataIndexNode.getEndOffset();
if (i != metadataIndexEntryList.size() - 1) {
endOffset = metadataIndexEntryList.get(i + 1).getOffset();
}
@@ -1459,6 +1460,86 @@ public class TsFileSequenceReader implements AutoCloseable {
}
/**
+ * Get AlignedChunkMetadata of sensors under one device
+ *
+ * @param device device name
+ */
+ public List<AlignedChunkMetadata> getAlignedChunkMetadata(String device) throws IOException {
+ readFileMetadata();
+ MetadataIndexNode deviceMetadataIndexNode = tsFileMetaData.getMetadataIndex();
+ Pair<MetadataIndexEntry, Long> metadataIndexPair =
+ getMetadataAndEndOffset(deviceMetadataIndexNode, device, true, true);
+ if (metadataIndexPair == null) {
+ throw new IOException("Device {" + device + "} is not in tsFileMetaData");
+ }
+ ByteBuffer buffer = readData(metadataIndexPair.left.getOffset(), metadataIndexPair.right);
+ MetadataIndexNode metadataIndexNode;
+ TimeseriesMetadata firstTimeseriesMetadata;
+ try {
+ // next layer MeasurementNode of the specific DeviceNode
+ metadataIndexNode = MetadataIndexNode.deserializeFrom(buffer);
+ } catch (BufferOverflowException e) {
+ logger.error(METADATA_INDEX_NODE_DESERIALIZE_ERROR, file);
+ throw e;
+ }
+ firstTimeseriesMetadata = tryToGetFirstTimeseriesMetadata(metadataIndexNode);
+ if (firstTimeseriesMetadata == null) {
+ throw new IOException("Timeseries of device {" + device + "} are not aligned");
+ }
+
+ Map<String, List<TimeseriesMetadata>> timeseriesMetadataMap = new TreeMap<>();
+ List<MetadataIndexEntry> metadataIndexEntryList = metadataIndexNode.getChildren();
+
+ for (int i = 0; i < metadataIndexEntryList.size(); i++) {
+ MetadataIndexEntry metadataIndexEntry = metadataIndexEntryList.get(i);
+ long endOffset = metadataIndexNode.getEndOffset();
+ if (i != metadataIndexEntryList.size() - 1) {
+ endOffset = metadataIndexEntryList.get(i + 1).getOffset();
+ }
+ buffer = readData(metadataIndexEntry.getOffset(), endOffset);
+ if (metadataIndexNode.getNodeType().equals(MetadataIndexNodeType.LEAF_MEASUREMENT)) {
+ List<TimeseriesMetadata> timeseriesMetadataList = new ArrayList<>();
+ while (buffer.hasRemaining()) {
+ timeseriesMetadataList.add(TimeseriesMetadata.deserializeFrom(buffer, true));
+ }
+ timeseriesMetadataMap
+ .computeIfAbsent(device, k -> new ArrayList<>())
+ .addAll(timeseriesMetadataList);
+ } else {
+ generateMetadataIndex(
+ metadataIndexEntry,
+ buffer,
+ device,
+ metadataIndexNode.getNodeType(),
+ timeseriesMetadataMap,
+ true);
+ }
+ }
+
+ for (List<TimeseriesMetadata> timeseriesMetadataList : timeseriesMetadataMap.values()) {
+ TimeseriesMetadata timeseriesMetadata = timeseriesMetadataList.get(0);
+ List<TimeseriesMetadata> valueTimeseriesMetadataList = new ArrayList<>();
+
+ for (int i = 1; i < timeseriesMetadataList.size(); i++) {
+ valueTimeseriesMetadataList.add(timeseriesMetadataList.get(i));
+ }
+
+ AlignedTimeSeriesMetadata alignedTimeSeriesMetadata =
+ new AlignedTimeSeriesMetadata(timeseriesMetadata, valueTimeseriesMetadataList);
+ List<AlignedChunkMetadata> chunkMetadataList = new ArrayList<>();
+ for (IChunkMetadata chunkMetadata : readIChunkMetaDataList(alignedTimeSeriesMetadata)) {
+ chunkMetadataList.add((AlignedChunkMetadata) chunkMetadata);
+ }
+ // only one timeseriesMetadataList in one device
+ return chunkMetadataList;
+ }
+
+ throw new IOException(
+ String.format(
+ "Error when reading timeseriesMetadata of device %s in file %s", device, file));
+ }
+
+ /**
* get ChunkMetaDatas in given TimeseriesMetaData
*
* @return List of ChunkMetaData
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/GetAlignedChunkMetadataTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/GetAlignedChunkMetadataTest.java
new file mode 100644
index 0000000..194a547
--- /dev/null
+++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/GetAlignedChunkMetadataTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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.tsfile.read;
+
+import org.apache.iotdb.tsfile.file.metadata.AlignedChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata;
+import org.apache.iotdb.tsfile.utils.TsFileGeneratorForTest;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+public class GetAlignedChunkMetadataTest {
+ @Test
+ public void getAlignedChunkMetadataTest() {
+ // generate aligned timeseries "d1.s1","d1.s2","d1.s3","d1.s4" and nonAligned timeseries
+ // "d2.s1","d2.s2","d2.s3"
+ TsFileGeneratorForTest.generateAlignedTsFile(10, 100, 30);
+ String filePath = TsFileGeneratorForTest.alignedOutputDataFile;
+ try (TsFileSequenceReader reader = new TsFileSequenceReader(filePath)) {
+ // query for non-exist device
+ try {
+ reader.getAlignedChunkMetadata("d3");
+ } catch (IOException e) {
+ Assert.assertEquals("Device {d3} is not in tsFileMetaData", e.getMessage());
+ }
+
+ // query for non-aligned device
+ try {
+ reader.getAlignedChunkMetadata("d2");
+ } catch (IOException e) {
+ Assert.assertEquals("Timeseries of device {d2} are not aligned", e.getMessage());
+ }
+
+ String[] expected = new String[] {"s1", "s2", "s3", "s4"};
+
+ List<AlignedChunkMetadata> chunkMetadataList = reader.getAlignedChunkMetadata("d1");
+ AlignedChunkMetadata alignedChunkMetadata = chunkMetadataList.get(0);
+ Assert.assertEquals("", alignedChunkMetadata.getTimeChunkMetadata().getMeasurementUid());
+ int i = 0;
+ for (IChunkMetadata chunkMetadata : alignedChunkMetadata.getValueChunkMetadataList()) {
+ Assert.assertEquals(expected[i], chunkMetadata.getMeasurementUid());
+ i++;
+ }
+
+ Assert.assertEquals(expected.length, i);
+
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ TsFileGeneratorForTest.closeAlignedTsFile();
+ }
+}