[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313297965 ## File path: tsfile/src/main/java/org/apache/iotdb/tsfile/read/controller/MetadataQuerier.java ## @@ -62,4 +62,9 @@ */ List convertSpace2TimePartition(List paths, long spacePartitionStartPos, long spacePartitionEndPos) throws IOException; + + /** + * clear caches (if used) to release memory. + */ + void clear(); Review comment: rename this class to IMetadataQuerier? This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313248755
##
File path: server/src/main/java/org/apache/iotdb/db/utils/TsPrimitiveType.java
##
@@ -74,6 +74,30 @@ public Binary getBinary() {
throw new UnsupportedOperationException("getBinary() is not supported for
current sub-class");
}
+ public void setBoolean(boolean val) {
+throw new UnsupportedOperationException("getBoolean() is not supported for
current sub-class");
+ }
+
+ public void setInt(int val) {
+throw new UnsupportedOperationException("getInt() is not supported for
current sub-class");
Review comment:
the same ...
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313734754
##
File path: server/src/main/java/org/apache/iotdb/db/utils/MergeUtils.java
##
@@ -0,0 +1,354 @@
+/**
+ * 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.utils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadataIndex;
+import org.apache.iotdb.tsfile.file.metadata.TsFileMetaData;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderWithoutFilter;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MergeUtils {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MergeUtils.class);
+
+ private MergeUtils() {
+// util class
+ }
+
+ public static void writeTVPair(TimeValuePair timeValuePair, IChunkWriter
chunkWriter) {
+switch (chunkWriter.getDataType()) {
+ case TEXT:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getBinary());
+break;
+ case DOUBLE:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getDouble());
+break;
+ case BOOLEAN:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getBoolean());
+break;
+ case INT64:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getLong());
+break;
+ case INT32:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getInt());
+break;
+ case FLOAT:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getFloat());
+break;
+ default:
+throw new UnsupportedOperationException("Unknown data type " +
chunkWriter.getDataType());
+}
+ }
+
+ /**
+ * Collect all paths contained in the all SeqFiles and UnseqFiles in a merge
and sort them
+ * before return.
+ * @param resource
+ * @return all paths contained in the merge.
+ * @throws IOException
+ */
+ public static List collectPaths(MergeResource resource)
+ throws IOException {
+Set pathSet = new HashSet<>();
+for (TsFileResource tsFileResource : resource.getUnseqFiles()) {
+ TsFileSequenceReader sequenceReader =
resource.getFileReader(tsFileResource);
+
resource.getMeasurementSchemaMap().putAll(sequenceReader.readFileMetadata().getMeasurementSchema());
+ pathSet.addAll(collectFileSeries(sequenceReader));
+}
+for (TsFileResource tsFileResource : resource.getSeqFiles()) {
+ TsFileSequenceReader sequenceReader =
resource.getFileReader(tsFileResource);
+
resource.getMeasurementSchemaMap().putAll(sequenceReader.readFileMetadata().getMeasurementSchema());
+ pathSet.addAll(collectFileSeries(sequenceReader));
+}
+List ret = new ArrayList<>(pathSet);
+ret.sort(Comparator.comparing(Path::getFullPath));
+return ret;
+ }
+
+ private static List collectFileSeries(TsFileSequenceReader
sequenceReader) throws IOException {
+TsFileMetaData metaData = sequenceReader.readFileMetadata();
+Set deviceIds = metaData.getDeviceMap().keySet();
+Set measurements = metaData.getMeasurementSchema().keySet();
+List paths = new ArrayList<>();
+for (String deviceId : deviceIds) {
+ for (String
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313248648
##
File path: server/src/main/java/org/apache/iotdb/db/utils/TsPrimitiveType.java
##
@@ -74,6 +74,30 @@ public Binary getBinary() {
throw new UnsupportedOperationException("getBinary() is not supported for
current sub-class");
}
+ public void setBoolean(boolean val) {
+throw new UnsupportedOperationException("getBoolean() is not supported for
current sub-class");
Review comment:
```suggestion
throw new UnsupportedOperationException("setBoolean() is not supported
for current sub-class");
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r314219509
##
File path:
server/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/CachedUnseqResourceMergeReader.java
##
@@ -0,0 +1,42 @@
+/**
+ * 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.query.reader.resourceRelated;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iotdb.db.query.reader.chunkRelated.CachedDiskChunkReader;
+import org.apache.iotdb.db.query.reader.universal.CachedPriorityMergeReader;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderWithoutFilter;
+
+public class CachedUnseqResourceMergeReader extends CachedPriorityMergeReader {
Review comment:
```suggestion
public class CachedUnseqFilesMergeReader extends CachedPriorityMergeReader {
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313685171
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##
@@ -185,53 +204,95 @@ public StorageGroupProcessor(String systemInfoDir,
String storageGroupName)
private void recover() throws ProcessorException {
logger.info("recover Storage Group {}", storageGroupName);
-// collect TsFiles from sequential data directory
-List tsFiles =
getAllFiles(DirectoryManager.getInstance().getAllSequenceFileFolders());
-recoverSeqFiles(tsFiles);
-
-// collect TsFiles from unsequential data directory
-tsFiles =
getAllFiles(DirectoryManager.getInstance().getAllUnSequenceFileFolders());
-recoverUnseqFiles(tsFiles);
+try {
+ // collect TsFiles from sequential and unsequential data directory
+ List seqTsFiles =
getAllFiles(DirectoryManager.getInstance().getAllSequenceFileFolders());
+ List unseqTsFiles =
+
getAllFiles(DirectoryManager.getInstance().getAllUnSequenceFileFolders());
+
+ recoverSeqFiles(seqTsFiles);
+ recoverUnseqFiles(unseqTsFiles);
+
+ String taskName = storageGroupName + "-" + System.currentTimeMillis();
+ File mergingMods = new File(storageGroupSysDir,
MERGING_MODIFICAITON_FILE_NAME);
+ if (mergingMods.exists()) {
+mergingModification = new ModificationFile(storageGroupSysDir +
File.separator + MERGING_MODIFICAITON_FILE_NAME);
Review comment:
How about changing the type of parameter from String to FIle?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313352319
##
File path:
tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/ForceAppendTsFileWriter.java
##
@@ -0,0 +1,87 @@
+/**
+ * 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.write.writer;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
+import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadata;
+import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadataIndex;
+import org.apache.iotdb.tsfile.file.metadata.TsFileMetaData;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+/**
+ * ForceAppendTsFileWriter opens a COMPLETE TsFile, reads and truncate its
metadata to support
+ * appending new data.
+ */
+public class ForceAppendTsFileWriter extends TsFileIOWriter{
+
+ private Map knownSchemas = new HashMap<>();
Review comment:
```suggestion
private Map knownSchemas;
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313829348
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxFileMergeFileSelector.java
##
@@ -0,0 +1,303 @@
+/**
+ * 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.engine.merge.selector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.MergeException;
+import org.apache.iotdb.db.utils.MergeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MaxFileMergeFileSelector selects the most files from given seqFiles and
unseqFiles which can be
+ * merged without exceeding given memory budget. It always assume the number
of timeseries being
+ * queried at the same time is 1 to maximize the number of file merged.
+ */
+public class MaxFileMergeFileSelector implements MergeFileSelector {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MaxFileMergeFileSelector.class);
+ private static final String LOG_FILE_COST = "Memory cost of file {} is {}";
+
+ MergeResource resource;
+
+ long totalCost;
+ private long memoryBudget;
+ private long maxSeqFileCost;
+
+ // the number of timeseries being queried at the same time
+ int concurrentMergeNum = 1;
+
+ /**
+ * Total metadata size of each file.
+ */
+ private Map fileMetaSizeMap = new HashMap<>();
+ /**
+ * Maximum memory cost of querying a timeseries in each file.
+ */
+ private Map maxSeriesQueryCostMap = new HashMap<>();
+
+ List selectedUnseqFiles;
+ List selectedSeqFiles;
+
+ private Collection tmpSelectedSeqFiles;
+ private long tempMaxSeqFileCost;
+
+ private boolean[] seqSelected;
+ private int seqSelectedNum;
+
+ public MaxFileMergeFileSelector(MergeResource resource, long memoryBudget) {
+this.resource = resource;
+this.memoryBudget = memoryBudget;
+ }
+
+ /**
+ * Select merge candidates from seqFiles and unseqFiles under the given
memoryBudget.
+ * This process iteratively adds the next unseqFile from unseqFiles and its
overlapping seqFiles
+ * as newly-added candidates and computes their estimated memory cost. If
the current cost
+ * pluses the new cost is still under the budget, accept the unseqFile and
the seqFiles as
+ * candidates, otherwise go to the next iteration.
+ * The memory cost of a file is calculated in two ways:
+ *The rough estimation: for a seqFile, the size of its metadata is used
for estimation.
+ *Since in the worst case, the file only contains one timeseries and all
its metadata will
+ *be loaded into memory with at most one actual data page (which is
negligible) and writing
+ *the timeseries into a new file generate metadata of the similar size,
so the size of all
+ *seqFiles' metadata (generated when writing new chunks) pluses the
largest one (loaded
+ *when reading a timeseries from the seqFiles) is the total estimation
of all seqFiles; for
+ *an unseqFile, since the merge reader may read all chunks of a series
to perform a merge
+ *read, the whole file may be loaded into memory, so we use the file's
length as the
+ *maximum estimation.
+ *The tight estimation: based on the rough estimation, we scan the
file's metadata to
+ *count the number of chunks for each series, find the series which have
the most
+ *chunks in the file and use its chunk proportion to refine the rough
estimation.
+ * The rough estimation is performed first, if no candidates can be found
using rough
+ * estimation, we run the selection again with tight estimation.
+ * @return two lists of TsFileResource, the former is selected seqFiles and
the latter is
+ * selected unseqFiles or an empty array if there are
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313736890
##
File path: server/src/main/java/org/apache/iotdb/db/utils/MergeUtils.java
##
@@ -0,0 +1,354 @@
+/**
+ * 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.utils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadataIndex;
+import org.apache.iotdb.tsfile.file.metadata.TsFileMetaData;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderWithoutFilter;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MergeUtils {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MergeUtils.class);
+
+ private MergeUtils() {
+// util class
+ }
+
+ public static void writeTVPair(TimeValuePair timeValuePair, IChunkWriter
chunkWriter) {
+switch (chunkWriter.getDataType()) {
+ case TEXT:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getBinary());
+break;
+ case DOUBLE:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getDouble());
+break;
+ case BOOLEAN:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getBoolean());
+break;
+ case INT64:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getLong());
+break;
+ case INT32:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getInt());
+break;
+ case FLOAT:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getFloat());
+break;
+ default:
+throw new UnsupportedOperationException("Unknown data type " +
chunkWriter.getDataType());
+}
+ }
+
+ /**
+ * Collect all paths contained in the all SeqFiles and UnseqFiles in a merge
and sort them
+ * before return.
+ * @param resource
+ * @return all paths contained in the merge.
+ * @throws IOException
+ */
+ public static List collectPaths(MergeResource resource)
+ throws IOException {
+Set pathSet = new HashSet<>();
+for (TsFileResource tsFileResource : resource.getUnseqFiles()) {
+ TsFileSequenceReader sequenceReader =
resource.getFileReader(tsFileResource);
+
resource.getMeasurementSchemaMap().putAll(sequenceReader.readFileMetadata().getMeasurementSchema());
+ pathSet.addAll(collectFileSeries(sequenceReader));
+}
+for (TsFileResource tsFileResource : resource.getSeqFiles()) {
+ TsFileSequenceReader sequenceReader =
resource.getFileReader(tsFileResource);
+
resource.getMeasurementSchemaMap().putAll(sequenceReader.readFileMetadata().getMeasurementSchema());
+ pathSet.addAll(collectFileSeries(sequenceReader));
+}
+List ret = new ArrayList<>(pathSet);
+ret.sort(Comparator.comparing(Path::getFullPath));
+return ret;
+ }
+
+ private static List collectFileSeries(TsFileSequenceReader
sequenceReader) throws IOException {
+TsFileMetaData metaData = sequenceReader.readFileMetadata();
+Set deviceIds = metaData.getDeviceMap().keySet();
+Set measurements = metaData.getMeasurementSchema().keySet();
+List paths = new ArrayList<>();
+for (String deviceId : deviceIds) {
+ for (String
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313823707
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
##
@@ -0,0 +1,261 @@
+/**
+ * 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.engine.merge.manage;
+
+import static org.apache.iotdb.db.engine.merge.task.MergeTask.MERGE_SUFFIX;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.reader.IPointReader;
+import
org.apache.iotdb.db.query.reader.resourceRelated.CachedUnseqResourceMergeReader;
+import org.apache.iotdb.db.utils.MergeUtils;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.write.chunk.ChunkBuffer;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MergeResource manages files and caches of readers, writers,
MeasurementSchemas and
+ * modifications to avoid unnecessary object creations and file openings.
+ */
+public class MergeResource {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MergeResource.class);
+
+ private List seqFiles;
+ private List unseqFiles;
+
+ private QueryContext mergeContext = new QueryContext();
+
+ private Map fileReaderCache;
+ private Map fileWriterCache = new
HashMap<>();
+ private Map> modificationCache = new
HashMap<>();
+ private Map measurementSchemaMap = new
HashMap<>();
+ private Map chunkWriterCache = new
ConcurrentHashMap<>();
+
+ private boolean cacheDeviceMeta = false;
+
+ public MergeResource(List seqFiles, List
unseqFiles) {
+this.seqFiles =
seqFiles.stream().filter(TsFileResource::isClosed).collect(Collectors.toList());
+this.unseqFiles =
+
unseqFiles.stream().filter(TsFileResource::isClosed).collect(Collectors.toList());
+this.fileReaderCache = new HashMap<>();
+ }
+
+ public void clear() throws IOException {
+for (TsFileSequenceReader sequenceReader : fileReaderCache.values()) {
+ sequenceReader.close();
+}
+
+fileReaderCache.clear();
+fileWriterCache.clear();
+modificationCache.clear();
+measurementSchemaMap.clear();
+chunkWriterCache.clear();
+ }
+
+ public MeasurementSchema getSchema(String measurementId) {
+return measurementSchemaMap.get(measurementId);
+ }
+
+ /**
+ * Construct a new or get an existing RestorableTsFileIOWriter of a merge
temp file for a
+ * SeqFile. The path of the merge temp file will be the seqFile's + ".merge".
+ * @param resource
+ * @return A RestorableTsFileIOWriter of a merge temp file for a SeqFile.
+ * @throws IOException
+ */
+ public RestorableTsFileIOWriter getMergeFileWriter(TsFileResource resource)
throws IOException {
+RestorableTsFileIOWriter writer = fileWriterCache.get(resource);
+if (writer == null) {
+ writer = new RestorableTsFileIOWriter(new
File(resource.getFile().getPath() + MERGE_SUFFIX));
+ fileWriterCache.put(resource, writer);
+}
+return writer;
+ }
+
+ /**
+ * Query ChunkMetadata of a timeseries from the given TsFile (seq or unseq).
The ChunkMetadata
+ * is not cached
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313687987
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/recover/MergeLogger.java
##
@@ -0,0 +1,142 @@
+/**
+ * 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.engine.merge.recover;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.List;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.read.common.Path;
+
+/**
+ * MergeLogger records the progress of a merge in file "merge.log" as text
lines.
+ */
+public class MergeLogger {
+
+ public static final String MERGE_LOG_NAME = "merge.log";
+
+ static final String STR_SEQ_FILES = "seqFiles";
+ static final String STR_UNSEQ_FILES = "unseqFiles";
+ static final String STR_TIMESERIES = "timeseries";
+ static final String STR_START = "start";
+ static final String STR_END = "end";
+ static final String STR_ALL_TS_END = "all ts end";
+ static final String STR_MERGE_START = "merge start";
+ static final String STR_MERGE_END = "merge end";
+
+ private BufferedWriter logStream;
+
+ public MergeLogger(String storageGroupDir) throws IOException {
+logStream = new BufferedWriter(new FileWriter(new File(storageGroupDir,
MERGE_LOG_NAME), true));
+ }
+
+ public void close() throws IOException {
+logStream.close();
+ }
+
+ public void logTSStart(List paths) throws IOException {
+logStream.write(STR_START);
+for (Path path : paths) {
+ logStream.write(" " + path.getFullPath());
+}
+logStream.newLine();
+logStream.flush();
+ }
+
+ public void logFilePositionUpdate(File file) throws IOException {
Review comment:
```suggestion
public void logFilePosition(File file) throws IOException {
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r314576210
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/recover/LogAnalyzer.java
##
@@ -0,0 +1,295 @@
+/**
+ * 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.engine.merge.recover;
+
+import static org.apache.iotdb.db.engine.merge.recover.MergeLogger.*;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.*;
+import java.util.Map.Entry;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * LogAnalyzer scans the "merge.log" file and recovers information such as
files of last merge,
+ * the last available positions of each file and how many timeseries and files
have been merged.
+ */
+public class LogAnalyzer {
+
+ private static final Logger logger =
LoggerFactory.getLogger(LogAnalyzer.class);
+
+ private MergeResource resource;
+ private String taskName;
+ private File logFile;
+
+ private Map fileLastPositions = new HashMap<>();
+ private Map tempFileLastPositions = new HashMap<>();
+
+ private List mergedPaths = new ArrayList<>();
+ private List unmergedPaths;
+ private List unmergedFiles;
+ private String currLine;
+
+ private Status status;
+
+ public LogAnalyzer(MergeResource resource, String taskName, File logFile) {
+this.resource = resource;
+this.taskName = taskName;
+this.logFile = logFile;
+ }
+
+ /**
+ * Scan through the logs to find out where the last merge has stopped and
store the information
+ * about the progress in the fields.
+ * @return a Status indicating the completed stage of the last merge.
+ * @throws IOException
+ */
+ public Status analyze() throws IOException {
+status = Status.NONE;
+try (BufferedReader bufferedReader = new BufferedReader(new
FileReader(logFile))) {
+ currLine = bufferedReader.readLine();
+ if (currLine != null) {
+analyzeSeqFiles(bufferedReader);
+
+analyzeUnseqFiles(bufferedReader);
+
+analyzeUnmergedSeries(bufferedReader);
+
+analyzeMergedSeries(bufferedReader, unmergedPaths);
+
+analyzeMergedFiles(bufferedReader);
+ }
+}
+return status;
+ }
+
+ private void analyzeUnmergedSeries(BufferedReader bufferedReader) throws
IOException {
+if (!STR_TIMESERIES.equals(currLine)) {
+ return;
+}
+long startTime = System.currentTimeMillis();
+List paths = new ArrayList<>();
+while ((currLine = bufferedReader.readLine()) != null) {
+ if (STR_MERGE_START.equals(currLine)) {
+break;
+ }
+ paths.add(new Path(currLine));
+}
+if (logger.isDebugEnabled()) {
+ logger.debug("{} found {} seq files after {}ms", taskName, paths.size(),
+ (System.currentTimeMillis() - startTime));
+}
+unmergedPaths = paths;
Review comment:
allPathsToBeMerged?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313364622 ## File path: server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java ## @@ -34,7 +35,7 @@ import org.slf4j.LoggerFactory; /** - * FileReaderManager is a singleton, which is used to manage + * resource.getSeqFiles()ager is a singleton, which is used to manage Review comment: what is ager? This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313370128
##
File path:
server/src/main/java/org/apache/iotdb/db/query/control/JobFileManager.java
##
@@ -91,13 +89,13 @@ void removeUsedFilesForGivenJob(long jobId) {
* so sealedFilePathsMap.get(jobId) or
unsealedFilePathsMap.get(jobId)
* must not return null.
*/
- void addFilePathToMap(long jobId, String filePath, boolean isSealed) {
-ConcurrentHashMap> pathMap = !isSealed ?
unsealedFilePathsMap :
+ void addFilePathToMap(long jobId, TsFileResource tsFile, boolean isSealed) {
Review comment:
```suggestion
void addFilePathToMap(long jobId, TsFileResource tsFile, boolean isClosed)
{
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313197710 ## File path: server/src/assembly/resources/conf/iotdb-engine.properties ## @@ -155,7 +155,52 @@ concurrent_flush_thread=0 # whether take over the memory management by IoTDB rather than JVM when serializing memtable as bytes in memory # (i.e., whether use ChunkBufferPool), value true, false -chunk_buffer_pool_enable = false +chunk_buffer_pool_enable=false + + +### Merge Configurations + + +# How many thread will be set up to perform merge main tasks, 1 by default. +# Set to 1 when less than or equal to 0. +merge_thread_num=1 + +# How many thread will be set up to perform merge chunk sub-tasks, 8 by default. +# Set to 1 when less than or equal to 0. +merge_chunk_subthread_num=8 + +# If one merge file selection runs for more than this time, it will be ended and its current +# selection will be used as final selection. Unit: millis. +# When < 0, it means time is unbounded. +merge_fileSelection_time_budget=3 + +# How much memory may be used in ONE merge task (in byte), 20% of maximum JVM memory by default. +# This is only a rough estimation, starting from a relatively small value to avoid OOM. +# Each new merge thread may take such memory, so merge_thread_num * merge_memory_budget is the +# total memory estimation of merge. +# merge_memory_budget=2147483648 + +# When set to true, if some crashed merges are detected during system rebooting, such merges will +# be continued, otherwise, the unfinished parts of such merges will not be continued while the +# finished parts still remains as they are. +# If you are feeling the rebooting is too slow, set this to false, false by default +continue_merge_after_reboot=false + +# A global merge will be performed each such interval, that is, each storage group will be merged +# (if proper merge candidates can be found). Unit: second, default: 1hours. +# When less than or equal to 0, timed merge is disabled. +merge_interval_sec=3600 + +# When set to true, all merges becomes full merge (the whole SeqFiles are re-written despite how +# much they are overflowed). This may increase merge overhead depending on how much the SeqFiles +# are overflowed. +force_full_merge=false + +# During a merge, if a chunk with less number of chunks than this parameter, the chunk will be Review comment: less number of points? This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r314206947
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
##
@@ -0,0 +1,261 @@
+/**
+ * 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.engine.merge.manage;
+
+import static org.apache.iotdb.db.engine.merge.task.MergeTask.MERGE_SUFFIX;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.reader.IPointReader;
+import
org.apache.iotdb.db.query.reader.resourceRelated.CachedUnseqResourceMergeReader;
+import org.apache.iotdb.db.utils.MergeUtils;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.write.chunk.ChunkBuffer;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MergeResource manages files and caches of readers, writers,
MeasurementSchemas and
+ * modifications to avoid unnecessary object creations and file openings.
+ */
+public class MergeResource {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MergeResource.class);
+
+ private List seqFiles;
+ private List unseqFiles;
+
+ private QueryContext mergeContext = new QueryContext();
+
+ private Map fileReaderCache;
+ private Map fileWriterCache = new
HashMap<>();
+ private Map> modificationCache = new
HashMap<>();
+ private Map measurementSchemaMap = new
HashMap<>();
+ private Map chunkWriterCache = new
ConcurrentHashMap<>();
+
+ private boolean cacheDeviceMeta = false;
+
+ public MergeResource(List seqFiles, List
unseqFiles) {
+this.seqFiles =
seqFiles.stream().filter(TsFileResource::isClosed).collect(Collectors.toList());
+this.unseqFiles =
+
unseqFiles.stream().filter(TsFileResource::isClosed).collect(Collectors.toList());
+this.fileReaderCache = new HashMap<>();
+ }
+
+ public void clear() throws IOException {
Review comment:
This method is not implemented as the name, it does not clear two resources
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313190255
##
File path: client/src/main/java/org/apache/iotdb/cli/client/AbstractClient.java
##
@@ -728,12 +728,10 @@ protected static void importCmd(String specialCmd,
String cmd, IoTDBConnection c
}
}
- protected static void executeQuery(IoTDBConnection connection, String cmd) {
-Statement statement = null;
+ private static void executeQuery(IoTDBConnection connection, String cmd) {
long startTime = System.currentTimeMillis();
-try {
+try (Statement statement = connection.createStatement();) {
Review comment:
```suggestion
try (Statement statement = connection.createStatement()) {
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313704705
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
##
@@ -0,0 +1,261 @@
+/**
+ * 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.engine.merge.manage;
+
+import static org.apache.iotdb.db.engine.merge.task.MergeTask.MERGE_SUFFIX;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.reader.IPointReader;
+import
org.apache.iotdb.db.query.reader.resourceRelated.CachedUnseqResourceMergeReader;
+import org.apache.iotdb.db.utils.MergeUtils;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.write.chunk.ChunkBuffer;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MergeResource manages files and caches of readers, writers,
MeasurementSchemas and
+ * modifications to avoid unnecessary object creations and file openings.
+ */
+public class MergeResource {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MergeResource.class);
+
+ private List seqFiles;
+ private List unseqFiles;
+
+ private QueryContext mergeContext = new QueryContext();
+
+ private Map fileReaderCache;
+ private Map fileWriterCache = new
HashMap<>();
+ private Map> modificationCache = new
HashMap<>();
+ private Map measurementSchemaMap = new
HashMap<>();
+ private Map chunkWriterCache = new
ConcurrentHashMap<>();
+
+ private boolean cacheDeviceMeta = false;
+
+ public MergeResource(List seqFiles, List
unseqFiles) {
+this.seqFiles =
seqFiles.stream().filter(TsFileResource::isClosed).collect(Collectors.toList());
+this.unseqFiles =
+
unseqFiles.stream().filter(TsFileResource::isClosed).collect(Collectors.toList());
+this.fileReaderCache = new HashMap<>();
+ }
+
+ public void clear() throws IOException {
+for (TsFileSequenceReader sequenceReader : fileReaderCache.values()) {
+ sequenceReader.close();
+}
+
+fileReaderCache.clear();
+fileWriterCache.clear();
+modificationCache.clear();
+measurementSchemaMap.clear();
+chunkWriterCache.clear();
+ }
+
+ public MeasurementSchema getSchema(String measurementId) {
+return measurementSchemaMap.get(measurementId);
+ }
+
+ /**
+ * Construct a new or get an existing RestorableTsFileIOWriter of a merge
temp file for a
+ * SeqFile. The path of the merge temp file will be the seqFile's + ".merge".
+ * @param resource
+ * @return A RestorableTsFileIOWriter of a merge temp file for a SeqFile.
+ * @throws IOException
+ */
+ public RestorableTsFileIOWriter getMergeFileWriter(TsFileResource resource)
throws IOException {
+RestorableTsFileIOWriter writer = fileWriterCache.get(resource);
+if (writer == null) {
+ writer = new RestorableTsFileIOWriter(new
File(resource.getFile().getPath() + MERGE_SUFFIX));
+ fileWriterCache.put(resource, writer);
+}
+return writer;
+ }
+
+ /**
+ * Query ChunkMetadata of a timeseries from the given TsFile (seq or unseq).
The ChunkMetadata
+ * is not cached
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313352890
##
File path:
tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/ForceAppendTsFileWriter.java
##
@@ -0,0 +1,87 @@
+/**
+ * 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.write.writer;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
+import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadata;
+import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadataIndex;
+import org.apache.iotdb.tsfile.file.metadata.TsFileMetaData;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+/**
+ * ForceAppendTsFileWriter opens a COMPLETE TsFile, reads and truncate its
metadata to support
+ * appending new data.
+ */
+public class ForceAppendTsFileWriter extends TsFileIOWriter{
+
+ private Map knownSchemas = new HashMap<>();
+ private long truncatePosition;
+
+ public ForceAppendTsFileWriter(File file) throws IOException {
+this.out = new DefaultTsFileOutput(file, true);
+this.file = file;
+
+// file doesn't exist
+if (file.length() == 0 || !file.exists()) {
+ throw new TsFileNotCompleteException("File " + file.getPath() + " is not
a complete TsFile");
+}
+
+try (TsFileSequenceReader reader = new
TsFileSequenceReader(file.getAbsolutePath(), true)) {
+
+ // this tsfile is not complete
+ if (!reader.isComplete()) {
+throw new TsFileNotCompleteException("File " + file.getPath() + " is
not a complete TsFile");
+ }
+ TsFileMetaData fileMetaData = reader.readFileMetadata();
+ Map deviceMap =
fileMetaData.getDeviceMap();
+ long firstDeviceMetaPos = Long.MAX_VALUE;
+ for (Entry deviceMetadataEntry :
deviceMap.entrySet()) {
Review comment:
```suggestion
for (TsDeviceMetadataIndex deviceMetadataIndex : deviceMap.values()) {
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313197861 ## File path: server/src/assembly/resources/conf/iotdb-engine.properties ## @@ -155,7 +155,52 @@ concurrent_flush_thread=0 # whether take over the memory management by IoTDB rather than JVM when serializing memtable as bytes in memory # (i.e., whether use ChunkBufferPool), value true, false -chunk_buffer_pool_enable = false +chunk_buffer_pool_enable=false + + +### Merge Configurations + + +# How many thread will be set up to perform merge main tasks, 1 by default. +# Set to 1 when less than or equal to 0. +merge_thread_num=1 + +# How many thread will be set up to perform merge chunk sub-tasks, 8 by default. +# Set to 1 when less than or equal to 0. +merge_chunk_subthread_num=8 + +# If one merge file selection runs for more than this time, it will be ended and its current +# selection will be used as final selection. Unit: millis. +# When < 0, it means time is unbounded. +merge_fileSelection_time_budget=3 + +# How much memory may be used in ONE merge task (in byte), 20% of maximum JVM memory by default. +# This is only a rough estimation, starting from a relatively small value to avoid OOM. +# Each new merge thread may take such memory, so merge_thread_num * merge_memory_budget is the +# total memory estimation of merge. +# merge_memory_budget=2147483648 + +# When set to true, if some crashed merges are detected during system rebooting, such merges will +# be continued, otherwise, the unfinished parts of such merges will not be continued while the +# finished parts still remains as they are. +# If you are feeling the rebooting is too slow, set this to false, false by default +continue_merge_after_reboot=false + +# A global merge will be performed each such interval, that is, each storage group will be merged +# (if proper merge candidates can be found). Unit: second, default: 1hours. +# When less than or equal to 0, timed merge is disabled. +merge_interval_sec=3600 + +# When set to true, all merges becomes full merge (the whole SeqFiles are re-written despite how +# much they are overflowed). This may increase merge overhead depending on how much the SeqFiles +# are overflowed. +force_full_merge=false + +# During a merge, if a chunk with less number of chunks than this parameter, the chunk will be +# merged with its succeeding chunks even if it is not overflowed, until the merged chunks reach +# this threshold and the new chunk will be flushed. +# When less than 0, this mechanism is disabled. +chunk_merge_point_threshold=4096 Review comment: by default, this is the point number in one page, would it be better if we increase this threshold? This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313208276 ## File path: server/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/UnseqResourceMergeReader.java ## @@ -120,6 +121,17 @@ public UnseqResourceMergeReader(Path seriesPath, List unseqResou } } + public UnseqResourceMergeReader(Path seriesPath, List chunks, Filter filter) Review comment: this method is unused, remove? This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313774688
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MergeFileSelector.java
##
@@ -0,0 +1,34 @@
+/**
+ * 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.engine.merge.selector;
+
+import java.util.List;
+import org.apache.iotdb.db.exception.MergeException;
+
+/**
+ * MergeFileSelector selects a set of files from given seqFiles and unseqFiles
which can be
+ * merged without exceeding given memory budget.
+ */
+public interface MergeFileSelector {
Review comment:
```suggestion
public interface IMergeFileSelector {
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r314209390
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##
@@ -674,6 +739,136 @@ public void closeUnsealedTsFileProcessor(
}
}
+ public void merge(boolean fullMerge) {
+writeLock();
+try {
+ if (isMerging) {
+if (logger.isInfoEnabled()) {
+ logger.info("{} Last merge is ongoing, currently consumed time:
{}ms", storageGroupName,
+ (System.currentTimeMillis() - mergeStartTime));
+}
+return;
+ }
+ if (unSequenceFileList.isEmpty() || sequenceFileList.isEmpty()) {
+logger.info("{} no files to be merged", storageGroupName);
+return;
+ }
+
+ long budget =
IoTDBDescriptor.getInstance().getConfig().getMergeMemoryBudget();
+ MergeResource mergeResource = new MergeResource(sequenceFileList,
unSequenceFileList);
+ MergeFileSelector fileSelector = getMergeFileSelector(budget,
mergeResource);
+ try {
+List[] mergeFiles = fileSelector.select();
+if (mergeFiles.length == 0) {
+ logger.info("{} cannot select merge candidates under the budget {}",
storageGroupName,
+ budget);
+ return;
+}
+// avoid pending tasks holds the metadata and streams
+mergeResource.clear();
+String taskName = storageGroupName + "-" + System.currentTimeMillis();
+// do not cache metadata until true candidates are chosen, or too much
metadata will be
+// cached during selection
+mergeResource.setCacheDeviceMeta(true);
+
+MergeTask mergeTask = new MergeTask(mergeResource,
storageGroupSysDir.getPath(),
+this::mergeEndAction, taskName, fullMerge,
fileSelector.getConcurrentMergeNum());
+mergingModification = new ModificationFile(storageGroupSysDir +
File.separator + MERGING_MODIFICAITON_FILE_NAME);
+MergeManager.getINSTANCE().submitMainTask(mergeTask);
+if (logger.isInfoEnabled()) {
+ logger.info("{} submits a merge task {}, merging {} seqFiles, {}
unseqFiles",
+ storageGroupName, taskName, mergeFiles[0].size(),
mergeFiles[1].size());
+}
+isMerging = true;
+mergeStartTime = System.currentTimeMillis();
+
+ } catch (MergeException | IOException e) {
+logger.error("{} cannot select file for merge", storageGroupName, e);
+ }
+} finally {
+ writeUnlock();
+}
+ }
+
+ private MergeFileSelector getMergeFileSelector(long budget, MergeResource
resource) {
+MergeFileStrategy strategy =
IoTDBDescriptor.getInstance().getConfig().getMergeFileStrategy();
+switch (strategy) {
+ case MAX_FILE_NUM:
+return new MaxFileMergeFileSelector(resource, budget);
+ case MAX_SERIES_NUM:
+return new MaxSeriesMergeFileSelector(resource, budget);
+ default:
+throw new UnsupportedOperationException("Unknown MergeFileStrategy " +
strategy);
+}
+ }
+
+ protected void mergeEndAction(List seqFiles,
List unseqFiles,
Review comment:
add javadoc
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313820437
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxFileMergeFileSelector.java
##
@@ -0,0 +1,303 @@
+/**
+ * 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.engine.merge.selector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.MergeException;
+import org.apache.iotdb.db.utils.MergeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MaxFileMergeFileSelector selects the most files from given seqFiles and
unseqFiles which can be
+ * merged without exceeding given memory budget. It always assume the number
of timeseries being
+ * queried at the same time is 1 to maximize the number of file merged.
+ */
+public class MaxFileMergeFileSelector implements MergeFileSelector {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MaxFileMergeFileSelector.class);
+ private static final String LOG_FILE_COST = "Memory cost of file {} is {}";
+
+ MergeResource resource;
+
+ long totalCost;
+ private long memoryBudget;
+ private long maxSeqFileCost;
+
+ // the number of timeseries being queried at the same time
+ int concurrentMergeNum = 1;
+
+ /**
+ * Total metadata size of each file.
+ */
+ private Map fileMetaSizeMap = new HashMap<>();
+ /**
+ * Maximum memory cost of querying a timeseries in each file.
+ */
+ private Map maxSeriesQueryCostMap = new HashMap<>();
+
+ List selectedUnseqFiles;
+ List selectedSeqFiles;
+
+ private Collection tmpSelectedSeqFiles;
+ private long tempMaxSeqFileCost;
+
+ private boolean[] seqSelected;
+ private int seqSelectedNum;
+
+ public MaxFileMergeFileSelector(MergeResource resource, long memoryBudget) {
+this.resource = resource;
+this.memoryBudget = memoryBudget;
+ }
+
+ /**
+ * Select merge candidates from seqFiles and unseqFiles under the given
memoryBudget.
+ * This process iteratively adds the next unseqFile from unseqFiles and its
overlapping seqFiles
+ * as newly-added candidates and computes their estimated memory cost. If
the current cost
+ * pluses the new cost is still under the budget, accept the unseqFile and
the seqFiles as
+ * candidates, otherwise go to the next iteration.
+ * The memory cost of a file is calculated in two ways:
+ *The rough estimation: for a seqFile, the size of its metadata is used
for estimation.
+ *Since in the worst case, the file only contains one timeseries and all
its metadata will
+ *be loaded into memory with at most one actual data page (which is
negligible) and writing
Review comment:
I think we read at least a chunk at a time
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313346240
##
File path:
tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
##
@@ -612,4 +646,38 @@ public long selfCheck(Map
newSchema,
// so that we can continue to write data into this tsfile.
return truncatedPosition;
}
+
+ public int getTotalChunkNum() {
+return totalChunkNum;
+ }
+
+ public List getChunkMetadata(Path path) throws IOException {
Review comment:
```suggestion
public List getChunkMetadataList(Path path) throws
IOException {
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r314575470
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/recover/LogAnalyzer.java
##
@@ -0,0 +1,295 @@
+/**
+ * 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.engine.merge.recover;
+
+import static org.apache.iotdb.db.engine.merge.recover.MergeLogger.*;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.*;
+import java.util.Map.Entry;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * LogAnalyzer scans the "merge.log" file and recovers information such as
files of last merge,
+ * the last available positions of each file and how many timeseries and files
have been merged.
+ */
+public class LogAnalyzer {
+
+ private static final Logger logger =
LoggerFactory.getLogger(LogAnalyzer.class);
+
+ private MergeResource resource;
+ private String taskName;
+ private File logFile;
+
+ private Map fileLastPositions = new HashMap<>();
+ private Map tempFileLastPositions = new HashMap<>();
+
+ private List mergedPaths = new ArrayList<>();
+ private List unmergedPaths;
+ private List unmergedFiles;
+ private String currLine;
+
+ private Status status;
+
+ public LogAnalyzer(MergeResource resource, String taskName, File logFile) {
+this.resource = resource;
+this.taskName = taskName;
+this.logFile = logFile;
+ }
+
+ /**
+ * Scan through the logs to find out where the last merge has stopped and
store the information
+ * about the progress in the fields.
+ * @return a Status indicating the completed stage of the last merge.
+ * @throws IOException
+ */
+ public Status analyze() throws IOException {
+status = Status.NONE;
+try (BufferedReader bufferedReader = new BufferedReader(new
FileReader(logFile))) {
+ currLine = bufferedReader.readLine();
+ if (currLine != null) {
+analyzeSeqFiles(bufferedReader);
+
+analyzeUnseqFiles(bufferedReader);
+
+analyzeUnmergedSeries(bufferedReader);
+
+analyzeMergedSeries(bufferedReader, unmergedPaths);
+
+analyzeMergedFiles(bufferedReader);
+ }
+}
+return status;
+ }
+
+ private void analyzeUnmergedSeries(BufferedReader bufferedReader) throws
IOException {
+if (!STR_TIMESERIES.equals(currLine)) {
+ return;
+}
+long startTime = System.currentTimeMillis();
+List paths = new ArrayList<>();
+while ((currLine = bufferedReader.readLine()) != null) {
+ if (STR_MERGE_START.equals(currLine)) {
+break;
+ }
+ paths.add(new Path(currLine));
+}
+if (logger.isDebugEnabled()) {
+ logger.debug("{} found {} seq files after {}ms", taskName, paths.size(),
+ (System.currentTimeMillis() - startTime));
+}
+unmergedPaths = paths;
+ }
+
+ private void analyzeSeqFiles(BufferedReader bufferedReader) throws
IOException {
+if (!STR_SEQ_FILES.equals(currLine)) {
+ return;
+}
+long startTime = System.currentTimeMillis();
+List mergeSeqFiles = new ArrayList<>();
+while ((currLine = bufferedReader.readLine()) != null) {
+ if (STR_UNSEQ_FILES.equals(currLine)) {
+break;
+ }
+ Iterator iterator = resource.getSeqFiles().iterator();
+ while (iterator.hasNext()) {
+TsFileResource seqFile = iterator.next();
+if (seqFile.getFile().getAbsolutePath().equals(currLine)) {
Review comment:
could this resource list be different from the log file?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313852348
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxFileMergeFileSelector.java
##
@@ -0,0 +1,303 @@
+/**
+ * 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.engine.merge.selector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.MergeException;
+import org.apache.iotdb.db.utils.MergeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MaxFileMergeFileSelector selects the most files from given seqFiles and
unseqFiles which can be
+ * merged without exceeding given memory budget. It always assume the number
of timeseries being
+ * queried at the same time is 1 to maximize the number of file merged.
+ */
+public class MaxFileMergeFileSelector implements MergeFileSelector {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MaxFileMergeFileSelector.class);
+ private static final String LOG_FILE_COST = "Memory cost of file {} is {}";
+
+ MergeResource resource;
+
+ long totalCost;
+ private long memoryBudget;
+ private long maxSeqFileCost;
+
+ // the number of timeseries being queried at the same time
+ int concurrentMergeNum = 1;
+
+ /**
+ * Total metadata size of each file.
+ */
+ private Map fileMetaSizeMap = new HashMap<>();
+ /**
+ * Maximum memory cost of querying a timeseries in each file.
+ */
+ private Map maxSeriesQueryCostMap = new HashMap<>();
+
+ List selectedUnseqFiles;
+ List selectedSeqFiles;
+
+ private Collection tmpSelectedSeqFiles;
+ private long tempMaxSeqFileCost;
+
+ private boolean[] seqSelected;
+ private int seqSelectedNum;
+
+ public MaxFileMergeFileSelector(MergeResource resource, long memoryBudget) {
+this.resource = resource;
+this.memoryBudget = memoryBudget;
+ }
+
+ /**
+ * Select merge candidates from seqFiles and unseqFiles under the given
memoryBudget.
+ * This process iteratively adds the next unseqFile from unseqFiles and its
overlapping seqFiles
+ * as newly-added candidates and computes their estimated memory cost. If
the current cost
+ * pluses the new cost is still under the budget, accept the unseqFile and
the seqFiles as
+ * candidates, otherwise go to the next iteration.
+ * The memory cost of a file is calculated in two ways:
+ *The rough estimation: for a seqFile, the size of its metadata is used
for estimation.
+ *Since in the worst case, the file only contains one timeseries and all
its metadata will
+ *be loaded into memory with at most one actual data page (which is
negligible) and writing
+ *the timeseries into a new file generate metadata of the similar size,
so the size of all
+ *seqFiles' metadata (generated when writing new chunks) pluses the
largest one (loaded
+ *when reading a timeseries from the seqFiles) is the total estimation
of all seqFiles; for
+ *an unseqFile, since the merge reader may read all chunks of a series
to perform a merge
+ *read, the whole file may be loaded into memory, so we use the file's
length as the
+ *maximum estimation.
+ *The tight estimation: based on the rough estimation, we scan the
file's metadata to
+ *count the number of chunks for each series, find the series which have
the most
+ *chunks in the file and use its chunk proportion to refine the rough
estimation.
+ * The rough estimation is performed first, if no candidates can be found
using rough
+ * estimation, we run the selection again with tight estimation.
+ * @return two lists of TsFileResource, the former is selected seqFiles and
the latter is
+ * selected unseqFiles or an empty array if there are
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313693421
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/recover/MergeLogger.java
##
@@ -0,0 +1,142 @@
+/**
+ * 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.engine.merge.recover;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.List;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.read.common.Path;
+
+/**
+ * MergeLogger records the progress of a merge in file "merge.log" as text
lines.
+ */
+public class MergeLogger {
Review comment:
It's better to give an example of merge.log. Suppose there are 3 sequence
files and 2 unsequence files including 2 timeseries, what will the merge.log
looks like at the end of the merge.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313691947
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/recover/MergeLogger.java
##
@@ -0,0 +1,142 @@
+/**
+ * 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.engine.merge.recover;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.List;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.read.common.Path;
+
+/**
+ * MergeLogger records the progress of a merge in file "merge.log" as text
lines.
+ */
+public class MergeLogger {
+
+ public static final String MERGE_LOG_NAME = "merge.log";
+
+ static final String STR_SEQ_FILES = "seqFiles";
+ static final String STR_UNSEQ_FILES = "unseqFiles";
+ static final String STR_TIMESERIES = "timeseries";
+ static final String STR_START = "start";
+ static final String STR_END = "end";
Review comment:
```suggestion
static final String STR_TS_END = "ts end";
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313686847
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/task/RecoverMergeTask.java
##
@@ -0,0 +1,267 @@
+/**
+ * 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.engine.merge.task;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map.Entry;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.recover.LogAnalyzer;
+import org.apache.iotdb.db.engine.merge.recover.LogAnalyzer.Status;
+import org.apache.iotdb.db.engine.merge.recover.MergeLogger;
+import org.apache.iotdb.db.engine.merge.selector.MaxSeriesMergeFileSelector;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.MergeUtils;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * RecoverMergeTask is an extension of MergeTask, which resumes the last merge
progress by
+ * scanning merge.log using LogAnalyzer and continue the unfinished merge.
+ */
+public class RecoverMergeTask extends MergeTask {
+
+ private static final Logger logger =
LoggerFactory.getLogger(RecoverMergeTask.class);
+
+ private LogAnalyzer analyzer;
+
+ public RecoverMergeTask(List seqFiles,
+ List unseqFiles, String storageGroupDir,
Review comment:
```suggestion
List unseqFiles, String storageGroupSysDir,
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313204795
##
File path:
server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
##
@@ -109,52 +109,6 @@ public long assignJobId() {
return jobId;
}
- /**
Review comment:
the JavaDoc of this class needs to be cleaned
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313716591
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeManager.java
##
@@ -0,0 +1,127 @@
+/**
+ * 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.engine.merge.manage;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.service.IService;
+import org.apache.iotdb.db.service.ServiceType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MergeManager provides a ThreadPool to queue and run all merge tasks to
restrain the total
+ * resources occupied by merge and manages a Timer to periodically issue a
global merge.
+ */
+public class MergeManager implements IService {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MergeManager.class);
+ private static final MergeManager INSTANCE = new MergeManager();
+
+ private AtomicInteger threadCnt = new AtomicInteger();
+ private ThreadPoolExecutor mergeTaskPool;
+ private ThreadPoolExecutor mergeChunkSubTaskPool;
+ private ScheduledExecutorService timedMergeThreadPool;
+
+ private MergeManager() {
+ }
+
+ public static MergeManager getINSTANCE() {
+return INSTANCE;
+ }
+
+ public void submitMainTask(MergeTask mergeTask) {
+mergeTaskPool.submit(mergeTask);
+ }
+
+ public Future submitChunkSubTask(Callable callable) {
+return mergeChunkSubTaskPool.submit(callable);
+ }
+
+ @Override
+ public void start() {
+if (mergeTaskPool == null) {
+ int threadNum =
IoTDBDescriptor.getInstance().getConfig().getMergeThreadNum();
+ if (threadNum <= 0) {
+threadNum = 1;
+ }
+
+ int chunkSubThreadNum =
IoTDBDescriptor.getInstance().getConfig().getMergeChunkSubThreadNum();
+ if (chunkSubThreadNum <= 0) {
+chunkSubThreadNum = 1;
+ }
+
+ mergeTaskPool =
+ (ThreadPoolExecutor) Executors.newFixedThreadPool(threadNum,
+ r -> new Thread(r, "MergeThread-" +
threadCnt.getAndIncrement()));
+ mergeChunkSubTaskPool =
+ (ThreadPoolExecutor) Executors.newFixedThreadPool(threadNum *
chunkSubThreadNum,
+ r -> new Thread(r, "MergeChunkSubThread-" +
threadCnt.getAndIncrement()));
+ long mergeInterval =
IoTDBDescriptor.getInstance().getConfig().getMergeIntervalSec();
+ if (mergeInterval > 0) {
+timedMergeThreadPool = Executors.newSingleThreadScheduledExecutor( r
-> new Thread(r,
+"TimedMergeThread"));
+timedMergeThreadPool.scheduleAtFixedRate(this::flushAll, mergeInterval,
+mergeInterval, TimeUnit.SECONDS);
+ }
+ logger.info("MergeManager started");
+}
+ }
+
+ @Override
+ public void stop() {
+if (mergeTaskPool != null) {
+ if (timedMergeThreadPool != null) {
+timedMergeThreadPool.shutdownNow();
+timedMergeThreadPool = null;
+ }
+ mergeTaskPool.shutdownNow();
+ mergeChunkSubTaskPool.shutdownNow();
+ logger.info("Waiting for task pool to shut down");
+ while (!mergeTaskPool.isTerminated() ||
!mergeChunkSubTaskPool.isTerminated() ) {
+// wait
+ }
+ mergeTaskPool = null;
+ logger.info("MergeManager stopped");
+}
+ }
+
+ @Override
+ public ServiceType getID() {
+return ServiceType.MERGE_SERVICE;
+ }
+
+ private void flushAll() {
Review comment:
```suggestion
private void mergeAll() {
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313735585
##
File path: server/src/main/java/org/apache/iotdb/db/utils/MergeUtils.java
##
@@ -0,0 +1,354 @@
+/**
+ * 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.utils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadataIndex;
+import org.apache.iotdb.tsfile.file.metadata.TsFileMetaData;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderWithoutFilter;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MergeUtils {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MergeUtils.class);
+
+ private MergeUtils() {
+// util class
+ }
+
+ public static void writeTVPair(TimeValuePair timeValuePair, IChunkWriter
chunkWriter) {
+switch (chunkWriter.getDataType()) {
+ case TEXT:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getBinary());
+break;
+ case DOUBLE:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getDouble());
+break;
+ case BOOLEAN:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getBoolean());
+break;
+ case INT64:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getLong());
+break;
+ case INT32:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getInt());
+break;
+ case FLOAT:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getFloat());
+break;
+ default:
+throw new UnsupportedOperationException("Unknown data type " +
chunkWriter.getDataType());
+}
+ }
+
+ /**
+ * Collect all paths contained in the all SeqFiles and UnseqFiles in a merge
and sort them
+ * before return.
+ * @param resource
+ * @return all paths contained in the merge.
+ * @throws IOException
+ */
+ public static List collectPaths(MergeResource resource)
+ throws IOException {
+Set pathSet = new HashSet<>();
+for (TsFileResource tsFileResource : resource.getUnseqFiles()) {
+ TsFileSequenceReader sequenceReader =
resource.getFileReader(tsFileResource);
+
resource.getMeasurementSchemaMap().putAll(sequenceReader.readFileMetadata().getMeasurementSchema());
+ pathSet.addAll(collectFileSeries(sequenceReader));
+}
+for (TsFileResource tsFileResource : resource.getSeqFiles()) {
+ TsFileSequenceReader sequenceReader =
resource.getFileReader(tsFileResource);
+
resource.getMeasurementSchemaMap().putAll(sequenceReader.readFileMetadata().getMeasurementSchema());
+ pathSet.addAll(collectFileSeries(sequenceReader));
+}
+List ret = new ArrayList<>(pathSet);
+ret.sort(Comparator.comparing(Path::getFullPath));
+return ret;
+ }
+
+ private static List collectFileSeries(TsFileSequenceReader
sequenceReader) throws IOException {
+TsFileMetaData metaData = sequenceReader.readFileMetadata();
+Set deviceIds = metaData.getDeviceMap().keySet();
+Set measurements = metaData.getMeasurementSchema().keySet();
+List paths = new ArrayList<>();
+for (String deviceId : deviceIds) {
+ for (String
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313353051
##
File path:
tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/ForceAppendTsFileWriter.java
##
@@ -0,0 +1,87 @@
+/**
+ * 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.write.writer;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
+import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadata;
+import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadataIndex;
+import org.apache.iotdb.tsfile.file.metadata.TsFileMetaData;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+/**
+ * ForceAppendTsFileWriter opens a COMPLETE TsFile, reads and truncate its
metadata to support
+ * appending new data.
+ */
+public class ForceAppendTsFileWriter extends TsFileIOWriter{
+
+ private Map knownSchemas = new HashMap<>();
+ private long truncatePosition;
+
+ public ForceAppendTsFileWriter(File file) throws IOException {
+this.out = new DefaultTsFileOutput(file, true);
+this.file = file;
+
+// file doesn't exist
+if (file.length() == 0 || !file.exists()) {
+ throw new TsFileNotCompleteException("File " + file.getPath() + " is not
a complete TsFile");
+}
+
+try (TsFileSequenceReader reader = new
TsFileSequenceReader(file.getAbsolutePath(), true)) {
+
+ // this tsfile is not complete
+ if (!reader.isComplete()) {
+throw new TsFileNotCompleteException("File " + file.getPath() + " is
not a complete TsFile");
+ }
+ TsFileMetaData fileMetaData = reader.readFileMetadata();
+ Map deviceMap =
fileMetaData.getDeviceMap();
+ long firstDeviceMetaPos = Long.MAX_VALUE;
+ for (Entry deviceMetadataEntry :
deviceMap.entrySet()) {
+TsDeviceMetadataIndex deviceMetadataIndex =
deviceMetadataEntry.getValue();
Review comment:
```suggestion
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313774935
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MergePathSelector.java
##
@@ -0,0 +1,32 @@
+/**
+ * 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.engine.merge.selector;
+
+import java.util.Iterator;
+import java.util.List;
+import org.apache.iotdb.tsfile.read.common.Path;
+
+/**
+ * MergePathSelector select paths to be merged at a time if all paths of a
device cannot be
+ * merged at the same time.
+ */
+public interface MergePathSelector extends Iterator> {
Review comment:
```suggestion
public interface IMergePathSelector extends Iterator> {
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313684907
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##
@@ -165,7 +184,7 @@ public StorageGroupProcessor(String systemInfoDir, String
storageGroupName)
this.fileSchema = constructFileSchema(storageGroupName);
try {
- File storageGroupSysDir = new File(systemInfoDir, storageGroupName);
+ storageGroupSysDir = new File(systemInfoDir, storageGroupName);
if (storageGroupSysDir.mkdirs()) {
logger.info("Storage Group system Directory {} doesn't exist, create
it",
storageGroupSysDir.getPath());
Review comment:
in the line 192, there is a typo of "create"
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313681792
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##
@@ -143,13 +158,17 @@
*/
@SuppressWarnings("unused") // to be used in merge
private ReentrantLock mergeDeleteLock = new ReentrantLock();
+ private ReentrantReadWriteLock mergeLock = new ReentrantReadWriteLock();
Review comment:
add javadoc?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313201751
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/cache/TsFileMetadataUtils.java
##
@@ -95,6 +97,9 @@ public static TsDeviceMetadata getTsDeviceMetaData(String
filePath, Path seriesP
}
}
}
+for (List chunkMetaDataList :
pathToChunkMetaDataList.values()) {
+
chunkMetaDataList.sort(Comparator.comparingLong(ChunkMetaData::getStartTime));
+}
Review comment:
is it because of merge that makes them out-of-order?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313736246
##
File path: server/src/main/java/org/apache/iotdb/db/utils/MergeUtils.java
##
@@ -0,0 +1,354 @@
+/**
+ * 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.utils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadataIndex;
+import org.apache.iotdb.tsfile.file.metadata.TsFileMetaData;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderWithoutFilter;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MergeUtils {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MergeUtils.class);
+
+ private MergeUtils() {
+// util class
+ }
+
+ public static void writeTVPair(TimeValuePair timeValuePair, IChunkWriter
chunkWriter) {
+switch (chunkWriter.getDataType()) {
+ case TEXT:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getBinary());
+break;
+ case DOUBLE:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getDouble());
+break;
+ case BOOLEAN:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getBoolean());
+break;
+ case INT64:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getLong());
+break;
+ case INT32:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getInt());
+break;
+ case FLOAT:
+chunkWriter.write(timeValuePair.getTimestamp(),
timeValuePair.getValue().getFloat());
+break;
+ default:
+throw new UnsupportedOperationException("Unknown data type " +
chunkWriter.getDataType());
+}
+ }
+
+ /**
+ * Collect all paths contained in the all SeqFiles and UnseqFiles in a merge
and sort them
+ * before return.
+ * @param resource
+ * @return all paths contained in the merge.
+ * @throws IOException
+ */
+ public static List collectPaths(MergeResource resource)
+ throws IOException {
+Set pathSet = new HashSet<>();
+for (TsFileResource tsFileResource : resource.getUnseqFiles()) {
+ TsFileSequenceReader sequenceReader =
resource.getFileReader(tsFileResource);
+
resource.getMeasurementSchemaMap().putAll(sequenceReader.readFileMetadata().getMeasurementSchema());
+ pathSet.addAll(collectFileSeries(sequenceReader));
+}
+for (TsFileResource tsFileResource : resource.getSeqFiles()) {
+ TsFileSequenceReader sequenceReader =
resource.getFileReader(tsFileResource);
+
resource.getMeasurementSchemaMap().putAll(sequenceReader.readFileMetadata().getMeasurementSchema());
+ pathSet.addAll(collectFileSeries(sequenceReader));
+}
+List ret = new ArrayList<>(pathSet);
+ret.sort(Comparator.comparing(Path::getFullPath));
+return ret;
+ }
+
+ private static List collectFileSeries(TsFileSequenceReader
sequenceReader) throws IOException {
+TsFileMetaData metaData = sequenceReader.readFileMetadata();
+Set deviceIds = metaData.getDeviceMap().keySet();
+Set measurements = metaData.getMeasurementSchema().keySet();
+List paths = new ArrayList<>();
+for (String deviceId : deviceIds) {
+ for (String
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313694613
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
##
@@ -0,0 +1,261 @@
+/**
+ * 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.engine.merge.manage;
+
+import static org.apache.iotdb.db.engine.merge.task.MergeTask.MERGE_SUFFIX;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.reader.IPointReader;
+import
org.apache.iotdb.db.query.reader.resourceRelated.CachedUnseqResourceMergeReader;
+import org.apache.iotdb.db.utils.MergeUtils;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.write.chunk.ChunkBuffer;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MergeResource manages files and caches of readers, writers,
MeasurementSchemas and
+ * modifications to avoid unnecessary object creations and file openings.
+ */
+public class MergeResource {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MergeResource.class);
+
+ private List seqFiles;
+ private List unseqFiles;
+
+ private QueryContext mergeContext = new QueryContext();
+
+ private Map fileReaderCache;
+ private Map fileWriterCache = new
HashMap<>();
+ private Map> modificationCache = new
HashMap<>();
+ private Map measurementSchemaMap = new
HashMap<>();
+ private Map chunkWriterCache = new
ConcurrentHashMap<>();
+
+ private boolean cacheDeviceMeta = false;
+
+ public MergeResource(List seqFiles, List
unseqFiles) {
+this.seqFiles =
seqFiles.stream().filter(TsFileResource::isClosed).collect(Collectors.toList());
+this.unseqFiles =
+
unseqFiles.stream().filter(TsFileResource::isClosed).collect(Collectors.toList());
+this.fileReaderCache = new HashMap<>();
+ }
+
+ public void clear() throws IOException {
+for (TsFileSequenceReader sequenceReader : fileReaderCache.values()) {
+ sequenceReader.close();
+}
+
+fileReaderCache.clear();
+fileWriterCache.clear();
Review comment:
Do RestorableTsFileIOWriters in fileWriterCache need to be closed?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r314202063
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##
@@ -185,53 +204,95 @@ public StorageGroupProcessor(String systemInfoDir,
String storageGroupName)
private void recover() throws ProcessorException {
logger.info("recover Storage Group {}", storageGroupName);
-// collect TsFiles from sequential data directory
-List tsFiles =
getAllFiles(DirectoryManager.getInstance().getAllSequenceFileFolders());
-recoverSeqFiles(tsFiles);
-
-// collect TsFiles from unsequential data directory
-tsFiles =
getAllFiles(DirectoryManager.getInstance().getAllUnSequenceFileFolders());
-recoverUnseqFiles(tsFiles);
+try {
+ // collect TsFiles from sequential and unsequential data directory
+ List seqTsFiles =
getAllFiles(DirectoryManager.getInstance().getAllSequenceFileFolders());
+ List unseqTsFiles =
+
getAllFiles(DirectoryManager.getInstance().getAllUnSequenceFileFolders());
+
+ recoverSeqFiles(seqTsFiles);
+ recoverUnseqFiles(unseqTsFiles);
+
+ String taskName = storageGroupName + "-" + System.currentTimeMillis();
+ File mergingMods = new File(storageGroupSysDir,
MERGING_MODIFICAITON_FILE_NAME);
+ if (mergingMods.exists()) {
+mergingModification = new ModificationFile(storageGroupSysDir +
File.separator + MERGING_MODIFICAITON_FILE_NAME);
+ }
+ RecoverMergeTask recoverMergeTask = new RecoverMergeTask(seqTsFiles,
unseqTsFiles,
+ storageGroupSysDir.getPath(), this::mergeEndAction, taskName,
+ IoTDBDescriptor.getInstance().getConfig().isForceFullMerge());
+ logger.info("{} a RecoverMergeTask {} starts...", storageGroupName,
taskName);
+
recoverMergeTask.recoverMerge(IoTDBDescriptor.getInstance().getConfig().isContinueMergeAfterReboot());
+ if
(!IoTDBDescriptor.getInstance().getConfig().isContinueMergeAfterReboot()) {
+mergingMods.delete();
+ }
+} catch (IOException e) {
+ throw new ProcessorException(e);
+}
for (TsFileResource resource : sequenceFileList) {
latestTimeForEachDevice.putAll(resource.getEndTimeMap());
latestFlushedTimeForEachDevice.putAll(resource.getEndTimeMap());
}
}
- private List getAllFiles(List folders) {
+ private List getAllFiles(List folders) throws
IOException {
List tsFiles = new ArrayList<>();
for (String baseDir : folders) {
File fileFolder = new File(baseDir, storageGroupName);
if (!fileFolder.exists()) {
continue;
}
+ // some TsFileResource may be persisting when the system crashed, try
recovering such
Review comment:
```suggestion
// some TsFileResource may be being persisted when the system crashed,
try recovering such
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313298216 ## File path: tsfile/src/main/java/org/apache/iotdb/tsfile/read/controller/ChunkLoader.java ## @@ -34,4 +34,8 @@ */ void close() throws IOException; + /** + * clear Chunk cache if used. + */ + void clear(); Review comment: rename this class to IChunkLoader? This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313705442
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
##
@@ -0,0 +1,261 @@
+/**
+ * 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.engine.merge.manage;
+
+import static org.apache.iotdb.db.engine.merge.task.MergeTask.MERGE_SUFFIX;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.reader.IPointReader;
+import
org.apache.iotdb.db.query.reader.resourceRelated.CachedUnseqResourceMergeReader;
+import org.apache.iotdb.db.utils.MergeUtils;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.write.chunk.ChunkBuffer;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MergeResource manages files and caches of readers, writers,
MeasurementSchemas and
+ * modifications to avoid unnecessary object creations and file openings.
+ */
+public class MergeResource {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MergeResource.class);
+
+ private List seqFiles;
+ private List unseqFiles;
+
+ private QueryContext mergeContext = new QueryContext();
Review comment:
this is unused now, remove ?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r314208495
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##
@@ -674,6 +739,136 @@ public void closeUnsealedTsFileProcessor(
}
}
+ public void merge(boolean fullMerge) {
+writeLock();
+try {
+ if (isMerging) {
+if (logger.isInfoEnabled()) {
+ logger.info("{} Last merge is ongoing, currently consumed time:
{}ms", storageGroupName,
+ (System.currentTimeMillis() - mergeStartTime));
+}
+return;
+ }
+ if (unSequenceFileList.isEmpty() || sequenceFileList.isEmpty()) {
+logger.info("{} no files to be merged", storageGroupName);
+return;
+ }
+
+ long budget =
IoTDBDescriptor.getInstance().getConfig().getMergeMemoryBudget();
+ MergeResource mergeResource = new MergeResource(sequenceFileList,
unSequenceFileList);
+ MergeFileSelector fileSelector = getMergeFileSelector(budget,
mergeResource);
+ try {
+List[] mergeFiles = fileSelector.select();
+if (mergeFiles.length == 0) {
+ logger.info("{} cannot select merge candidates under the budget {}",
storageGroupName,
+ budget);
+ return;
+}
+// avoid pending tasks holds the metadata and streams
+mergeResource.clear();
+String taskName = storageGroupName + "-" + System.currentTimeMillis();
+// do not cache metadata until true candidates are chosen, or too much
metadata will be
+// cached during selection
+mergeResource.setCacheDeviceMeta(true);
+
+MergeTask mergeTask = new MergeTask(mergeResource,
storageGroupSysDir.getPath(),
+this::mergeEndAction, taskName, fullMerge,
fileSelector.getConcurrentMergeNum());
+mergingModification = new ModificationFile(storageGroupSysDir +
File.separator + MERGING_MODIFICAITON_FILE_NAME);
+MergeManager.getINSTANCE().submitMainTask(mergeTask);
+if (logger.isInfoEnabled()) {
+ logger.info("{} submits a merge task {}, merging {} seqFiles, {}
unseqFiles",
+ storageGroupName, taskName, mergeFiles[0].size(),
mergeFiles[1].size());
+}
+isMerging = true;
+mergeStartTime = System.currentTimeMillis();
+
+ } catch (MergeException | IOException e) {
+logger.error("{} cannot select file for merge", storageGroupName, e);
+ }
+} finally {
+ writeUnlock();
+}
+ }
+
+ private MergeFileSelector getMergeFileSelector(long budget, MergeResource
resource) {
+MergeFileStrategy strategy =
IoTDBDescriptor.getInstance().getConfig().getMergeFileStrategy();
+switch (strategy) {
+ case MAX_FILE_NUM:
+return new MaxFileMergeFileSelector(resource, budget);
+ case MAX_SERIES_NUM:
+return new MaxSeriesMergeFileSelector(resource, budget);
+ default:
+throw new UnsupportedOperationException("Unknown MergeFileStrategy " +
strategy);
+}
+ }
+
+ protected void mergeEndAction(List seqFiles,
List unseqFiles,
+ File mergeLog) {
+logger.info("{} a merge task is ending...", storageGroupName);
+
+if (unseqFiles.isEmpty()) {
+ // merge runtime exception arose, just end this merge
+ isMerging = false;
+ logger.info("{} a merge task abnormally ends", storageGroupName);
+ return;
+}
+
+mergeLock.writeLock().lock();
+try {
+ unSequenceFileList.removeAll(unseqFiles);
+} finally {
+ mergeLock.writeLock().unlock();
+}
+
+for (int i = 0; i < unseqFiles.size(); i++) {
+ TsFileResource unseqFile = unseqFiles.get(i);
+ unseqFile.getMergeQueryLock().writeLock().lock();
+ try {
+unseqFile.remove();
+unseqFiles.remove(unseqFile);
Review comment:
is this needed?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request #258: [IOTDB-143]Development of merge
qiaojialin commented on a change in pull request #258: [IOTDB-143]Development
of merge
URL: https://github.com/apache/incubator-iotdb/pull/258#discussion_r313691907
##
File path:
server/src/main/java/org/apache/iotdb/db/engine/merge/recover/MergeLogger.java
##
@@ -0,0 +1,142 @@
+/**
+ * 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.engine.merge.recover;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.List;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.read.common.Path;
+
+/**
+ * MergeLogger records the progress of a merge in file "merge.log" as text
lines.
+ */
+public class MergeLogger {
+
+ public static final String MERGE_LOG_NAME = "merge.log";
+
+ static final String STR_SEQ_FILES = "seqFiles";
+ static final String STR_UNSEQ_FILES = "unseqFiles";
+ static final String STR_TIMESERIES = "timeseries";
+ static final String STR_START = "start";
Review comment:
```suggestion
static final String STR_TS_START = "ts start";
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
