This is an automated email from the ASF dual-hosted git repository. sunzesong pushed a commit to branch jira_936 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3c55742dfb8cba2622bc1f67bb4e15241a26a68d Author: samperson1997 <[email protected]> AuthorDate: Thu Oct 8 10:45:54 2020 +0800 Refactor getAllDevices in TsFileSequenceReader --- .../iotdb/tsfile/read/TsFileSequenceReader.java | 104 +++++++++++---------- .../iotdb/tsfile/read/GetAllDevicesTest.java | 88 +++++++++++++++++ .../tsfile/read/TsFileSequenceReaderTest.java | 1 - .../iotdb/tsfile/read/reader/ReaderTest.java | 6 +- .../apache/iotdb/tsfile/utils/FileGenerator.java | 79 +++++++++++++--- 5 files changed, 212 insertions(+), 66 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 b9a8226..776fb08 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 @@ -24,7 +24,6 @@ import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Comparator; import java.util.HashMap; -import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Map.Entry; @@ -33,6 +32,7 @@ import java.util.TreeMap; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.stream.Collectors; import org.apache.iotdb.tsfile.common.conf.TSFileConfig; import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; import org.apache.iotdb.tsfile.compress.IUnCompressor; @@ -96,7 +96,7 @@ public class TsFileSequenceReader implements AutoCloseable { /** * construct function for TsFileSequenceReader. * - * @param file -given file name + * @param file -given file name * @param loadMetadataSize -whether load meta data size */ public TsFileSequenceReader(String file, boolean loadMetadataSize) throws IOException { @@ -137,7 +137,7 @@ public class TsFileSequenceReader implements AutoCloseable { /** * construct function for TsFileSequenceReader. * - * @param input -given input + * @param input -given input * @param loadMetadataSize -load meta data size */ public TsFileSequenceReader(TsFileInput input, boolean loadMetadataSize) throws IOException { @@ -155,10 +155,10 @@ public class TsFileSequenceReader implements AutoCloseable { /** * construct function for TsFileSequenceReader. * - * @param input the input of a tsfile. The current position should be a markder and then a chunk - * Header, rather than the magic number - * @param fileMetadataPos the position of the file metadata in the TsFileInput from the beginning - * of the input to the current position + * @param input the input of a tsfile. The current position should be a markder and + * then a chunk Header, rather than the magic number + * @param fileMetadataPos the position of the file metadata in the TsFileInput from the beginning + * of the input to the current position * @param fileMetadataSize the byte size of the file metadata in the input */ public TsFileSequenceReader(TsFileInput input, long fileMetadataPos, int fileMetadataSize) { @@ -423,30 +423,31 @@ public class TsFileSequenceReader implements AutoCloseable { } private List<String> getAllDevices(MetadataIndexNode metadataIndexNode) throws IOException { - Set<String> deviceSet = new HashSet<>(); + List<String> deviceList = new ArrayList<>(); int metadataIndexListSize = metadataIndexNode.getChildren().size(); - for (int i = 0; i < metadataIndexListSize; i++) { - MetadataIndexEntry metadataIndex = metadataIndexNode.getChildren().get(i); - switch (metadataIndexNode.getNodeType()) { - case LEAF_MEASUREMENT: - case INTERNAL_MEASUREMENT: - for (MetadataIndexEntry index : metadataIndexNode.getChildren()) { - deviceSet.add(index.getName()); - } - break; - case LEAF_DEVICE: - case INTERNAL_DEVICE: - long endOffset = metadataIndexNode.getEndOffset(); - if (i != metadataIndexListSize - 1) { - endOffset = metadataIndexNode.getChildren().get(i + 1).getOffset(); - } - ByteBuffer buffer = readData(metadataIndex.getOffset(), endOffset); - MetadataIndexNode node = MetadataIndexNode.deserializeFrom(buffer); - deviceSet.addAll(getAllDevices(node)); - break; + if (metadataIndexNode.getNodeType().equals(MetadataIndexNodeType.INTERNAL_MEASUREMENT)) { + for (MetadataIndexEntry index : metadataIndexNode.getChildren()) { + deviceList.add(index.getName()); + } + } else { + for (int i = 0; i < metadataIndexListSize; i++) { + long endOffset = metadataIndexNode.getEndOffset(); + if (i != metadataIndexListSize - 1) { + endOffset = metadataIndexNode.getChildren().get(i + 1).getOffset(); + } + ByteBuffer buffer = readData(metadataIndexNode.getChildren().get(i).getOffset(), endOffset); + MetadataIndexNode node = MetadataIndexNode.deserializeFrom(buffer); + if (node.getNodeType().equals(MetadataIndexNodeType.LEAF_DEVICE)) { + // if node in next level is LEAF_DEVICE, put all devices in node entry into the set + deviceList.addAll(node.getChildren().stream().map(MetadataIndexEntry::getName).collect( + Collectors.toList())); + } else { + // keep traversing + deviceList.addAll(getAllDevices(node)); + } } } - return new ArrayList<>(deviceSet); + return deviceList; } /** @@ -508,9 +509,9 @@ public class TsFileSequenceReader implements AutoCloseable { /** * Traverse the metadata index from MetadataIndexEntry to get TimeseriesMetadatas * - * @param metadataIndex MetadataIndexEntry - * @param buffer byte buffer - * @param deviceId String + * @param metadataIndex MetadataIndexEntry + * @param buffer byte buffer + * @param deviceId String * @param timeseriesMetadataMap map: deviceId -> timeseriesMetadata list */ private void generateMetadataIndex(MetadataIndexEntry metadataIndex, ByteBuffer buffer, @@ -584,12 +585,13 @@ public class TsFileSequenceReader implements AutoCloseable { * Get target MetadataIndexEntry and its end offset * * @param metadataIndex given MetadataIndexNode - * @param name target device / measurement name - * @param type target MetadataIndexNodeType, either INTERNAL_DEVICE or INTERNAL_MEASUREMENT. When - * searching for a device node, return when it is not INTERNAL_DEVICE. Likewise, when searching - * for a measurement node, return when it is not INTERNAL_MEASUREMENT. This works for the - * situation when the index tree does NOT have the device level and ONLY has the measurement - * level. + * @param name target device / measurement name + * @param type target MetadataIndexNodeType, either INTERNAL_DEVICE or + * INTERNAL_MEASUREMENT. When searching for a device node, return when it is + * not INTERNAL_DEVICE. Likewise, when searching for a measurement node, + * return when it is not INTERNAL_MEASUREMENT. This works for the situation + * when the index tree does NOT have the device level and ONLY has the + * measurement level. * @return target MetadataIndexEntry, endOffset pair */ private Pair<MetadataIndexEntry, Long> getMetadataAndEndOffset(MetadataIndexNode metadataIndex, @@ -616,7 +618,7 @@ public class TsFileSequenceReader implements AutoCloseable { /** * read data from current position of the input, and deserialize it to a CHUNK_GROUP_FOOTER. * - * @param position the offset of the chunk group footer in the file + * @param position the offset of the chunk group footer in the file * @param markerRead true if the offset does not contains the marker , otherwise false * @return a CHUNK_GROUP_FOOTER * @throws IOException io error @@ -649,9 +651,9 @@ public class TsFileSequenceReader implements AutoCloseable { /** * read the chunk's header. * - * @param position the file offset of this chunk's header + * @param position the file offset of this chunk's header * @param chunkHeaderSize the size of chunk's header - * @param markerRead true if the offset does not contains the marker , otherwise false + * @param markerRead true if the offset does not contains the marker , otherwise false */ private ChunkHeader readChunkHeader(long position, int chunkHeaderSize, boolean markerRead) throws IOException { @@ -756,8 +758,8 @@ public class TsFileSequenceReader implements AutoCloseable { * changed. * * @param position the start position of data in the tsFileInput, or the current position if - * position = -1 - * @param size the size of data that want to read + * position = -1 + * @param size the size of data that want to read * @return data that been read. */ private ByteBuffer readData(long position, int size) throws IOException { @@ -783,8 +785,8 @@ public class TsFileSequenceReader implements AutoCloseable { * position. * * @param start the start position of data in the tsFileInput, or the current position if position - * = -1 - * @param end the end position of data that want to read + * = -1 + * @param end the end position of data that want to read * @return data that been read. */ private ByteBuffer readData(long start, long end) throws IOException { @@ -801,11 +803,11 @@ public class TsFileSequenceReader implements AutoCloseable { /** * Self Check the file and return the position before where the data is safe. * - * @param newSchema the schema on each time series in the file + * @param newSchema the schema on each time series in the file * @param chunkGroupMetadataList ChunkGroupMetadata List - * @param versionInfo version pair List - * @param fastFinish if true and the file is complete, then newSchema and chunkGroupMetadataList - * parameter will be not modified. + * @param versionInfo version pair List + * @param fastFinish if true and the file is complete, then newSchema and + * chunkGroupMetadataList parameter will be not modified. * @return the position of the file that is fine. All data after the position in the file should * be truncated. */ @@ -992,7 +994,7 @@ public class TsFileSequenceReader implements AutoCloseable { * get device names which has valid chunks in [start, end) * * @param start start of the partition - * @param end end of the partition + * @param end end of the partition * @return device names in range */ public List<String> getDeviceNameInRange(long start, long end) throws IOException { @@ -1010,8 +1012,8 @@ public class TsFileSequenceReader implements AutoCloseable { * Check if the device has at least one Chunk in this partition * * @param seriesMetadataMap chunkMetaDataList of each measurement - * @param start the start position of the space partition - * @param end the end position of the space partition + * @param start the start position of the space partition + * @param end the end position of the space partition */ private boolean hasDataInPartition(Map<String, List<ChunkMetadata>> seriesMetadataMap, long start, long end) { diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/GetAllDevicesTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/GetAllDevicesTest.java new file mode 100644 index 0000000..1ece3af --- /dev/null +++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/GetAllDevicesTest.java @@ -0,0 +1,88 @@ +/* + * 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 java.io.IOException; +import java.util.List; +import org.apache.iotdb.tsfile.common.conf.TSFileConfig; +import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; +import org.apache.iotdb.tsfile.utils.FileGenerator; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +public class GetAllDevicesTest { + + private final TSFileConfig conf = TSFileDescriptor.getInstance().getConfig(); + private int maxDegreeOfIndexNode; + private static final String FILE_PATH = FileGenerator.outputDataFile; + + @Before + public void before() { + maxDegreeOfIndexNode = conf.getMaxDegreeOfIndexNode(); + conf.setMaxDegreeOfIndexNode(3); + } + + @After + public void after() throws IOException { + FileGenerator.after(); + conf.setMaxDegreeOfIndexNode(maxDegreeOfIndexNode); + } + + @Test + public void testGetAllDevices1() throws IOException { + testGetAllDevices(2, 2); + } + + @Test + public void testGetAllDevices2() throws IOException { + testGetAllDevices(2, 50); + } + + @Test + public void testGetAllDevices3() throws IOException { + testGetAllDevices(50, 2); + } + + @Test + public void testGetAllDevices4() throws IOException { + testGetAllDevices(50, 50); + } + + public void testGetAllDevices(int deviceNum, int measurementNum) throws IOException { + FileGenerator.generateFile(10000, deviceNum, measurementNum); + TsFileSequenceReader fileReader = new TsFileSequenceReader(FILE_PATH); + ReadOnlyTsFile tsFile = new ReadOnlyTsFile(fileReader); + + // test + TsFileSequenceReader reader = new TsFileSequenceReader(FILE_PATH); + List<String> devices = reader.getAllDevices(); + Assert.assertEquals(deviceNum, devices.size()); + for (int i = 0; i < deviceNum; i++) { + Assert.assertTrue(devices.contains("d" + i)); + } + + // after + tsFile.close(); + FileGenerator.after(); + } + +} diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java index c1c5d18..b4c5671 100644 --- a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java +++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java @@ -29,7 +29,6 @@ import org.apache.iotdb.tsfile.file.MetaMarker; import org.apache.iotdb.tsfile.file.footer.ChunkGroupFooter; import org.apache.iotdb.tsfile.file.header.ChunkHeader; import org.apache.iotdb.tsfile.file.header.PageHeader; -import org.apache.iotdb.tsfile.file.metadata.TsFileMetadata; import org.apache.iotdb.tsfile.utils.FileGenerator; import org.apache.iotdb.tsfile.utils.Pair; import org.junit.After; diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/reader/ReaderTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/reader/ReaderTest.java index 431b0e7..63b2f39 100644 --- a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/reader/ReaderTest.java +++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/reader/ReaderTest.java @@ -46,10 +46,12 @@ public class ReaderTest { private static final String FILE_PATH = TsFileGeneratorForTest.outputDataFile; private TsFileSequenceReader fileReader; private MetadataQuerierByFileImpl metadataQuerierByFile; - private int rowCount = 1000000; + private int maxDegreeOfIndexNode; + private final int rowCount = 1000000; @Before public void before() throws IOException { + maxDegreeOfIndexNode = TSFileDescriptor.getInstance().getConfig().getMaxDegreeOfIndexNode(); TSFileDescriptor.getInstance().getConfig().setTimeEncoder("TS_2DIFF"); TSFileDescriptor.getInstance().getConfig().setMaxDegreeOfIndexNode(3); TsFileGeneratorForTest.generateFile(rowCount, 10 * 1024 * 1024, 10000); @@ -60,7 +62,7 @@ public class ReaderTest { @After public void after() throws IOException { fileReader.close(); - TSFileDescriptor.getInstance().getConfig().setMaxDegreeOfIndexNode(1024); + TSFileDescriptor.getInstance().getConfig().setMaxDegreeOfIndexNode(maxDegreeOfIndexNode); TsFileGeneratorForTest.after(); } diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FileGenerator.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FileGenerator.java index 2c14c57..e2e1b56 100755 --- a/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FileGenerator.java +++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FileGenerator.java @@ -22,6 +22,7 @@ import java.io.File; import java.io.FileNotFoundException; import java.io.FileWriter; import java.io.IOException; +import java.nio.file.Files; import java.util.Scanner; import org.apache.iotdb.tsfile.common.conf.TSFileConfig; import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; @@ -32,8 +33,8 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; import org.apache.iotdb.tsfile.read.common.Path; import org.apache.iotdb.tsfile.write.TsFileWriter; import org.apache.iotdb.tsfile.write.record.TSRecord; -import org.apache.iotdb.tsfile.write.schema.Schema; import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import org.apache.iotdb.tsfile.write.schema.Schema; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -47,10 +48,10 @@ public class FileGenerator { private static TsFileWriter innerWriter; private static String inputDataFile; private static String errorOutputDataFile; + private static final TSFileConfig config = TSFileDescriptor.getInstance().getConfig(); public static void generateFile(int rowCount, int maxNumberOfPointsInPage) throws IOException { ROW_COUNT = rowCount; - TSFileConfig config = TSFileDescriptor.getInstance().getConfig(); int oldMaxNumberOfPointsInPage = config.getMaxNumberOfPointsInPage(); config.setMaxNumberOfPointsInPage(maxNumberOfPointsInPage); @@ -59,6 +60,17 @@ public class FileGenerator { config.setMaxNumberOfPointsInPage(oldMaxNumberOfPointsInPage); } + public static void generateFile(int maxNumberOfPointsInPage, int deviceNum, + int measurementNum) throws IOException { + ROW_COUNT = 1; + int oldMaxNumberOfPointsInPage = config.getMaxNumberOfPointsInPage(); + config.setMaxNumberOfPointsInPage(maxNumberOfPointsInPage); + + prepare(deviceNum, measurementNum); + write(); + config.setMaxNumberOfPointsInPage(oldMaxNumberOfPointsInPage); + } + public static void generateFile() throws IOException { generateFile(1000, 10); } @@ -70,6 +82,13 @@ public class FileGenerator { generateSampleInputDataFile(); } + public static void prepare(int deviceNum, int measurementNum) throws IOException { + inputDataFile = TestConstant.BASE_OUTPUT_PATH.concat("perTestInputData"); + errorOutputDataFile = TestConstant.BASE_OUTPUT_PATH.concat("perTestErrorOutputData.tsfile"); + generateTestSchema(deviceNum, measurementNum); + generateSampleInputDataFile(deviceNum, measurementNum); + } + public static void after() { File file = new File(inputDataFile); if (file.exists()) { @@ -143,6 +162,27 @@ public class FileGenerator { fw.close(); } + static private void generateSampleInputDataFile(int deviceNum, int measurementNum) + throws IOException { + File file = new File(inputDataFile); + if (file.exists()) { + Files.delete(file.toPath()); + } + if (!file.getParentFile().mkdirs()) { + LOG.info("Failed to create file folder {}", file.getParentFile()); + } + FileWriter fw = new FileWriter(file); + + long startTime = 1480562618000L; + for (int i = 0; i < deviceNum; i++) { + for (int j = 0; j < measurementNum; j++) { + String d = "d" + i + "," + startTime + ",s" + j + "," + 1; + fw.write(d + "\r\n"); + } + } + fw.close(); + } + static public void write() throws IOException { File file = new File(outputDataFile); File errorFile = new File(errorOutputDataFile); @@ -153,7 +193,7 @@ public class FileGenerator { errorFile.delete(); } - innerWriter = new TsFileWriter(file, schema, TSFileDescriptor.getInstance().getConfig()); + innerWriter = new TsFileWriter(file, schema, config); // write try { @@ -166,13 +206,15 @@ public class FileGenerator { private static void generateTestSchema() { schema = new Schema(); - TSFileConfig conf = TSFileDescriptor.getInstance().getConfig(); schema.registerTimeseries(new Path("d1", "s1"), - new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.valueOf(conf.getValueEncoder()))); + new MeasurementSchema("s1", TSDataType.INT32, + TSEncoding.valueOf(config.getValueEncoder()))); schema.registerTimeseries(new Path("d1", "s2"), - new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.valueOf(conf.getValueEncoder()))); + new MeasurementSchema("s2", TSDataType.INT64, + TSEncoding.valueOf(config.getValueEncoder()))); schema.registerTimeseries(new Path("d1", "s3"), - new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.valueOf(conf.getValueEncoder()))); + new MeasurementSchema("s3", TSDataType.INT64, + TSEncoding.valueOf(config.getValueEncoder()))); schema.registerTimeseries(new Path("d1", "s4"), new MeasurementSchema("s4", TSDataType.TEXT, TSEncoding.PLAIN)); schema.registerTimeseries(new Path("d1", "s5"), @@ -182,15 +224,29 @@ public class FileGenerator { schema.registerTimeseries(new Path("d1", "s7"), new MeasurementSchema("s7", TSDataType.DOUBLE, TSEncoding.RLE)); schema.registerTimeseries(new Path("d2", "s1"), - new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.valueOf(conf.getValueEncoder()))); + new MeasurementSchema("s1", TSDataType.INT32, + TSEncoding.valueOf(config.getValueEncoder()))); schema.registerTimeseries(new Path("d2", "s2"), - new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.valueOf(conf.getValueEncoder()))); + new MeasurementSchema("s2", TSDataType.INT64, + TSEncoding.valueOf(config.getValueEncoder()))); schema.registerTimeseries(new Path("d2", "s3"), - new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.valueOf(conf.getValueEncoder()))); + new MeasurementSchema("s3", TSDataType.INT64, + TSEncoding.valueOf(config.getValueEncoder()))); schema.registerTimeseries(new Path("d2", "s4"), new MeasurementSchema("s4", TSDataType.TEXT, TSEncoding.PLAIN)); } + private static void generateTestSchema(int deviceNum, int measurementNum) { + schema = new Schema(); + for (int i = 0; i < deviceNum; i++) { + for (int j = 0; j < measurementNum; j++) { + schema.registerTimeseries(new Path("d" + i, "s" + j), + new MeasurementSchema("s" + j, TSDataType.INT32, + TSEncoding.valueOf(config.getValueEncoder()))); + } + } + } + private static void writeToTsFile(Schema schema) throws IOException, WriteProcessException { Scanner in = getDataFile(inputDataFile); long lineCount = 0; @@ -215,8 +271,7 @@ public class FileGenerator { static private Scanner getDataFile(String path) { File file = new File(path); try { - Scanner in = new Scanner(file); - return in; + return new Scanner(file); } catch (FileNotFoundException e) { e.printStackTrace(); return null;
