You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ch...@apache.org on 2020/11/25 01:02:44 UTC
[iotdb] branch fileIndex created (now 1e67eac)
This is an automated email from the ASF dual-hosted git repository.
chaow pushed a change to branch fileIndex
in repository https://gitbox.apache.org/repos/asf/iotdb.git.
at 1e67eac refactor the index
This branch includes the following new commits:
new 3077da8 [IOTDB-698] add time indexer interface
new f6c752d [IOTDB-698] add time indexer interface
new e4b39e6 fix conflict
new ba2ecda rename IndexerManager to FileIndexerManager
new 1e67eac refactor the index
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 03/05: fix conflict
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chaow pushed a commit to branch fileIndex
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e4b39e68755c3264e357a9a740d40f8bc1e22041
Author: chaow <xu...@gmail.com>
AuthorDate: Mon Sep 7 15:53:29 2020 +0800
fix conflict
---
.../resources/conf/iotdb-engine.properties | 7 ++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 8 ++
.../iotdb/db/engine/merge/task/MergeFileTask.java | 11 +-
.../db/engine/merge/task/RecoverMergeTask.java | 5 +-
.../engine/storagegroup/StorageGroupProcessor.java | 88 ++++++++-------
.../db/engine/storagegroup/TsFileProcessor.java | 24 ++--
.../db/engine/storagegroup/TsFileResource.java | 2 -
.../apache/iotdb/db/timeIndex/IndexerManager.java | 53 +++++----
.../{device/DeviceIndex.java => TimeIndex.java} | 16 +--
.../org/apache/iotdb/db/timeIndex/TimeIndexer.java | 27 ++++-
.../timeIndex/{device => }/UpdateIndexsParam.java | 16 +--
.../db/timeIndex/device/DeviceTimeIndexer.java | 124 ---------------------
.../db/timeIndex/device/LoadAllTimeIndexer.java | 20 ++--
.../timeIndex/device/RocksDBDeviceTimeIndexer.java | 20 ++--
14 files changed, 182 insertions(+), 239 deletions(-)
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties b/server/src/assembly/resources/conf/iotdb-engine.properties
index aae3444..4b33bfc 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -229,6 +229,7 @@ write_read_schema_free_memory_proportion=4:3:1:2
# primitive array size (length of each array) in array pool
primitive_array_size=128
+<<<<<<< HEAD
# Ratio of write memory for invoking flush disk, 0.4 by default
flush_proportion=0.4
@@ -253,6 +254,12 @@ max_waiting_time_when_insert_blocked=0
# estimated metadata size (in byte) of one timeseries in Mtree
estimated_series_size=300
+=======
+# device time indexer
+enable_device_indexer=false
+# 0, indexer base on file; 1, indexer base on rocksdb
+device_indexer_type=0
+>>>>>>> fix conflict
####################
### Upgrade Configurations
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 570272a..8b559db 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -517,6 +517,14 @@ public class IoTDBDescriptor {
conf.setDebugState(Boolean.parseBoolean(properties
.getProperty("debug_state", String.valueOf(conf.isDebugOn()))));
+ conf.setEnableDeviceIndexer((Boolean.parseBoolean(
+ properties.getProperty(
+ "enable_device_indexer", String.valueOf(conf.isEnableDeviceIndexer())))));
+
+ conf.setDeviceIndexerType((Integer.parseInt(
+ properties.getProperty(
+ "device_indexer_type", String.valueOf(conf.getDeviceIndexerType())))));
+
// mqtt
if (properties.getProperty(IoTDBConstant.MQTT_HOST_NAME) != null) {
conf.setMqttHost(properties.getProperty(IoTDBConstant.MQTT_HOST_NAME));
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
index 42486ca..ac645d6 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
@@ -38,12 +38,11 @@ import org.apache.iotdb.db.engine.merge.manage.MergeContext;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
import org.apache.iotdb.db.engine.merge.recover.MergeLogger;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.query.control.FileReaderManager;
-import org.apache.iotdb.db.timeIndex.device.DeviceIndex;
-import org.apache.iotdb.db.timeIndex.device.DeviceTimeIndexer;
import org.apache.iotdb.db.timeIndex.IndexerManager;
-import org.apache.iotdb.db.timeIndex.device.UpdateIndexsParam;
+import org.apache.iotdb.db.timeIndex.TimeIndexer;
import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
@@ -85,7 +84,7 @@ class MergeFileTask {
this.unmergedFiles = unmergedSeqFiles;
}
- void mergeFiles() throws IOException {
+ void mergeFiles() throws IOException, IllegalPathException {
// decide whether to write the unmerged chunks to the merge files or to move the merged chunks
// back to the origin seqFile's
if (logger.isInfoEnabled()) {
@@ -206,8 +205,8 @@ class MergeFileTask {
if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
// add new device index
// may Indexer need delete old index background
- DeviceTimeIndexer deviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(seqFile.getStorageGroupName());
- deviceTimeIndexer.addIndexForDevices(seqFile.getDeviceToIndexMap(), seqFile.getStartTimes(), seqFile.getEndTimes(), seqFile.getTsFilePath());
+ TimeIndexer timeIndexer = IndexerManager.getInstance().getSeqIndexer(seqFile.getStorageGroupName());
+ timeIndexer.addIndexForPaths(seqFile.getDeviceToIndexMap(), seqFile.getStartTimes(), seqFile.getEndTimes(), seqFile.getTsFilePath());
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/RecoverMergeTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/RecoverMergeTask.java
index 1607797..cec9aea 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/RecoverMergeTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/RecoverMergeTask.java
@@ -33,6 +33,7 @@ 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.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.utils.MergeUtils;
@@ -94,7 +95,7 @@ public class RecoverMergeTask extends MergeTask {
}
}
- private void resumeAfterFilesLogged(boolean continueMerge) throws IOException {
+ private void resumeAfterFilesLogged(boolean continueMerge) throws IOException, IllegalPathException {
if (continueMerge) {
resumeMergeProgress();
calculateConcurrentSeriesNum();
@@ -115,7 +116,7 @@ public class RecoverMergeTask extends MergeTask {
cleanUp(continueMerge);
}
- private void resumeAfterAllTsMerged(boolean continueMerge) throws IOException {
+ private void resumeAfterAllTsMerged(boolean continueMerge) throws IOException, IllegalPathException {
if (continueMerge) {
resumeMergeProgress();
MergeFileTask mergeFileTask = new MergeFileTask(taskName, mergeContext, mergeLogger, resource,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index 90b3769..11dcf16 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -83,7 +83,7 @@ import org.apache.iotdb.db.query.control.QueryFileManager;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.service.UpgradeSevice;
import org.apache.iotdb.db.timeIndex.IndexerManager;
-import org.apache.iotdb.db.timeIndex.device.DeviceTimeIndexer;
+import org.apache.iotdb.db.timeIndex.TimeIndexer;
import org.apache.iotdb.db.utils.CopyOnReadLinkedList;
import org.apache.iotdb.db.writelog.recover.TsFileRecoverPerformer;
import org.apache.iotdb.rpc.RpcUtils;
@@ -270,8 +270,8 @@ public class StorageGroupProcessor {
* unseqDeviceTimeIndexer manage the device Index of unseq tsfiles
*
*/
- private DeviceTimeIndexer seqDeviceTimeIndexer;
- private DeviceTimeIndexer unseqDeviceTimeIndexer;
+ private TimeIndexer seqTimeIndexer;
+ private TimeIndexer unseqTimeIndexer;
public StorageGroupProcessor(String systemDir, String storageGroupName,
TsFileFlushPolicy fileFlushPolicy) throws StorageGroupProcessorException {
@@ -387,11 +387,15 @@ public class StorageGroupProcessor {
if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
// init TimeIndexer
- seqDeviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(storageGroupName);
- unseqDeviceTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(storageGroupName);
- if (seqDeviceTimeIndexer != null) {
- seqDeviceTimeIndexer.init();
- unseqDeviceTimeIndexer.init();
+ try {
+ seqTimeIndexer = IndexerManager.getInstance().getSeqIndexer(storageGroupName);
+ unseqTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(storageGroupName);
+ if (seqTimeIndexer != null) {
+ seqTimeIndexer.init();
+ unseqTimeIndexer.init();
+ }
+ } catch (IllegalPathException e) {
+ throw new StorageGroupProcessorException(e);
}
}
@@ -1300,21 +1304,19 @@ public class StorageGroupProcessor {
tsFileManagement.remove(resource, isSeq);
if (isSeq) {
if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- DeviceTimeIndexer deviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(resource.getStorageGroupName());
- if (deviceTimeIndexer != null) {
- deviceTimeIndexer.deleteIndexForDevices(resource.getDeviceToIndexMap(), resource.getStartTimes(),
- resource.getEndTimes(), resource.getTsFilePath());
- }
+ TimeIndexer timeIndexer = IndexerManager.getInstance().getSeqIndexer(resource.getStorageGroupName());
+ timeIndexer.deleteIndexForPaths(resource.getDeviceToIndexMap(), resource.getStartTimes(),
+ resource.getEndTimes(), resource.getTsFilePath());
}
} else {
if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- DeviceTimeIndexer deviceTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(resource.getStorageGroupName());
- if (deviceTimeIndexer != null) {
- deviceTimeIndexer.deleteIndexForDevices(resource.getDeviceToIndexMap(), resource.getStartTimes(),
+ TimeIndexer timeIndexer = IndexerManager.getInstance().getUnseqIndexer(resource.getStorageGroupName());
+ timeIndexer.deleteIndexForPaths(resource.getDeviceToIndexMap(), resource.getStartTimes(),
resource.getEndTimes(), resource.getTsFilePath());
- }
}
}
+ } catch (IllegalPathException e) {
+ logger.error("Fail to get DeviceTimeIndexer for storage group {}, err:{}", resource.getStorageGroupName(), e.getMessage());
} finally {
resource.writeUnlock();
}
@@ -1402,8 +1404,8 @@ public class StorageGroupProcessor {
List<TsFileResource> unseqResources;
try {
if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- seqResources = seqDeviceTimeIndexer.filterByOneDevice(deviceId, timeFilter);
- unseqResources = unseqDeviceTimeIndexer.filterByOneDevice(deviceId, timeFilter);
+ seqResources = seqTimeIndexer.filterByPath(deviceId, timeFilter);
+ unseqResources = unseqTimeIndexer.filterByPath(deviceId, timeFilter);
List<TsFileResource> unsealedSeqFiles = getUnSealedListResourceForQuery(
tsFileManagement.getTsFileList(true),
deviceId, measurementId, context, timeFilter, true);
@@ -1414,7 +1416,7 @@ public class StorageGroupProcessor {
unsealedSeqFiles.addAll(unsealedUnseqFiles);
} else {
seqResources = getFileResourceListForQuery(tsFileManagement.getTsFileList(true),
- upgradeSeqFileList, deviceId, measurementId, context, timeFilter, true);
+ upgradeSeqFileList, deviceId, measurementId, context, timeFilter, true);
unseqResources = getFileResourceListForQuery(tsFileManagement.getTsFileList(false),
upgradeUnseqFileList, deviceId, measurementId, context, timeFilter, false);
}
@@ -1894,16 +1896,16 @@ public class StorageGroupProcessor {
updateLatestTimeMap(newTsFileResource);
}
if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- DeviceTimeIndexer deviceTimeIndexer = null;
+ TimeIndexer timeIndexer = null;
if (newTsFileResource.isSeq()) {
- deviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(newTsFileResource.getStorageGroupName());
+ timeIndexer = IndexerManager.getInstance().getSeqIndexer(newTsFileResource.getStorageGroupName());
} else {
- deviceTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
+ timeIndexer = IndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
}
- deviceTimeIndexer.addIndexForDevices(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
- newTsFileResource.getEndTimes(), newTsFileResource.getTsFilePath());
+ timeIndexer.addIndexForPaths(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
+ newTsFileResource.getEndTimes(), newTsFileResource.getTsFilePath());
}
- } catch (DiskSpaceInsufficientException e) {
+ } catch (DiskSpaceInsufficientException | IllegalPathException e) {
logger.error(
"Failed to append the tsfile {} to storage group processor {} because the disk space is insufficient.",
tsfileToBeInserted.getAbsolutePath(), tsfileToBeInserted.getParentFile().getName());
@@ -1964,14 +1966,20 @@ public class StorageGroupProcessor {
}
if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- DeviceTimeIndexer deviceTimeIndexer = null;
- if (newTsFileResource.isSeq()) {
- deviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(newTsFileResource.getStorageGroupName());
- } else {
- deviceTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
+ try {
+ TimeIndexer timeIndexer = null;
+ if (newTsFileResource.isSeq()) {
+ timeIndexer = IndexerManager.getInstance().getSeqIndexer(newTsFileResource.getStorageGroupName());
+ } else {
+ timeIndexer = IndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
+ }
+ timeIndexer.addIndexForPaths(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
+ newTsFileResource.getEndTimes(), newTsFileResource.getTsFilePath());
+ } catch (IllegalPathException e) {
+ logger.error("Fail to get DeviceTimeIndexer for storage group {}, err:{}", newTsFileResource.getStorageGroupName(), e.getMessage());
+ IoTDBDescriptor.getInstance().getConfig().setReadOnly(true);
+ throw new LoadFileException(e);
}
- deviceTimeIndexer.addIndexForDevices(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
- newTsFileResource.getEndTimes(), newTsFileResource.getTsFilePath());
}
// update latest time map
@@ -2375,17 +2383,21 @@ public class StorageGroupProcessor {
try {
tsFileResourceToBeDeleted.remove();
if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- DeviceTimeIndexer deviceTimeIndexer = null;
+ try {
+ TimeIndexer timeIndexer = null;
if (tsFileResourceToBeDeleted.isSeq()) {
- deviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
+ timeIndexer = IndexerManager.getInstance().getSeqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
} else {
- deviceTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
+ timeIndexer = IndexerManager.getInstance().getUnseqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
}
- if (deviceTimeIndexer != null) {
- deviceTimeIndexer.deleteIndexForDevices(tsFileResourceToBeDeleted.getDeviceToIndexMap(),
+ if (timeIndexer != null) {
+ timeIndexer.deleteIndexForPaths(tsFileResourceToBeDeleted.getDeviceToIndexMap(),
tsFileResourceToBeDeleted.getStartTimes(), tsFileResourceToBeDeleted.getEndTimes(),
tsFileResourceToBeDeleted.getTsFilePath());
}
+ } catch (IllegalPathException e) {
+ return false;
+ }
}
logger.info("Delete tsfile {} successfully.", tsFileResourceToBeDeleted.getTsFile());
} finally {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 8711bd5..d44c281 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -50,7 +50,11 @@ import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.UpdateEndTi
import org.apache.iotdb.db.engine.version.VersionController;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.WriteProcessException;
+<<<<<<< HEAD
import org.apache.iotdb.db.exception.metadata.MetadataException;
+=======
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+>>>>>>> fix conflict
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
@@ -59,8 +63,8 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.rescon.PrimitiveArrayManager;
import org.apache.iotdb.db.rescon.SystemInfo;
import org.apache.iotdb.db.utils.MemUtils;
-import org.apache.iotdb.db.timeIndex.device.DeviceTimeIndexer;
import org.apache.iotdb.db.timeIndex.IndexerManager;
+import org.apache.iotdb.db.timeIndex.TimeIndexer;
import org.apache.iotdb.db.utils.QueryUtils;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.iotdb.db.writelog.WALFlushListener;
@@ -816,12 +820,18 @@ public class TsFileProcessor {
long closeStartTime = System.currentTimeMillis();
if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
// update device index
- if (sequence) {
- DeviceTimeIndexer seqIndexer = IndexerManager.getInstance().getSeqIndexer(storageGroupName);
- seqIndexer.addIndexForDevices(tsFileResource.deviceToIndex, tsFileResource.startTimes, tsFileResource.endTimes, tsFileResource.getTsFilePath());
- } else {
- DeviceTimeIndexer unseqIndexer = IndexerManager.getInstance().getUnseqIndexer(storageGroupName);
- unseqIndexer.addIndexForDevices(tsFileResource.deviceToIndex, tsFileResource.startTimes, tsFileResource.endTimes, tsFileResource.getTsFilePath());
+ try {
+ TimeIndexer timeIndexer = null;
+ if (sequence) {
+ timeIndexer = IndexerManager.getInstance().getSeqIndexer(storageGroupName);
+ } else {
+ timeIndexer = IndexerManager.getInstance().getUnseqIndexer(storageGroupName);
+ }
+ timeIndexer.addIndexForPaths(tsFileResource.deviceToIndex, tsFileResource.startTimes,
+ tsFileResource.endTimes, tsFileResource.getTsFilePath());
+ } catch (IllegalPathException e) {
+ logger.error("Failed to endFile {} for storage group {}, err:{}", tsFileResource.getTsFile(),
+ tsFileResource.getStorageGroupName(), e.getMessage());
}
}
tsFileResource.serialize();
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index 66d42d9..ca2b819 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -46,7 +46,6 @@ import org.apache.iotdb.db.engine.upgrade.UpgradeTask;
import org.apache.iotdb.db.exception.PartitionViolationException;
import org.apache.iotdb.db.rescon.CachedStringPool;
import org.apache.iotdb.db.service.UpgradeSevice;
-import org.apache.iotdb.db.timeIndex.device.DeviceTimeIndexer;
import org.apache.iotdb.db.utils.FilePathUtils;
import org.apache.iotdb.db.utils.UpgradeUtils;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
@@ -167,7 +166,6 @@ public class TsFileResource {
*/
private long minPlanIndex = Long.MAX_VALUE;
- private DeviceTimeIndexer deviceTimeIndexer;
private String storageGroupName;
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java
index 504dfd4..eb09ab8 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java
@@ -21,7 +21,6 @@ package org.apache.iotdb.db.timeIndex;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.PartialPath;
-import org.apache.iotdb.db.timeIndex.device.DeviceTimeIndexer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -32,8 +31,8 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
public class IndexerManager {
private String indexerFilePath;
- private Map<PartialPath, DeviceTimeIndexer> seqIndexers;
- private Map<PartialPath, DeviceTimeIndexer> unseqIndexers;
+ private Map<PartialPath, TimeIndexer> seqIndexers;
+ private Map<PartialPath, TimeIndexer> unseqIndexers;
private ReentrantReadWriteLock lock;
private static final Logger logger = LoggerFactory.getLogger(IndexerManager.class);
@@ -69,31 +68,43 @@ public class IndexerManager {
return true;
}
- public void addSeqIndexer(PartialPath storageGroup, DeviceTimeIndexer deviceTimeIndexer) {
+ public void addSeqIndexer(PartialPath storageGroup, TimeIndexer TimeIndexer) {
lock.writeLock().lock();
- seqIndexers.put(storageGroup, deviceTimeIndexer);
- lock.writeLock().unlock();
+ try {
+ seqIndexers.put(storageGroup, TimeIndexer);
+ } finally {
+ lock.writeLock().unlock();
+ }
}
- public void addUnseqIndexer(PartialPath storageGroup, DeviceTimeIndexer deviceTimeIndexer) {
+ public void addUnseqIndexer(PartialPath storageGroup, TimeIndexer TimeIndexer) {
lock.writeLock().lock();
- unseqIndexers.put(storageGroup, deviceTimeIndexer);
- lock.writeLock().unlock();
+ try {
+ unseqIndexers.put(storageGroup, TimeIndexer);
+ } finally {
+ lock.writeLock().unlock();
+ }
}
public void deleteSeqIndexer(PartialPath storageGroup) {
lock.writeLock().lock();
- seqIndexers.remove(storageGroup);
- lock.writeLock().unlock();
+ try {
+ seqIndexers.remove(storageGroup);
+ } finally {
+ lock.writeLock().unlock();
+ }
}
public void deleteUnseqIndexer(PartialPath storageGroup) {
lock.writeLock().lock();
- unseqIndexers.remove(storageGroup);
- lock.writeLock().unlock();
+ try {
+ unseqIndexers.remove(storageGroup);
+ } finally {
+ lock.writeLock().unlock();
+ }
}
- public DeviceTimeIndexer getSeqIndexer(PartialPath storageGroup) {
+ public TimeIndexer getSeqIndexer(PartialPath storageGroup) {
lock.readLock().lock();
try {
return seqIndexers.get(storageGroup);
@@ -102,13 +113,13 @@ public class IndexerManager {
}
}
- public DeviceTimeIndexer getSeqIndexer(String storageGroup) {
+ public TimeIndexer getSeqIndexer(String storageGroup) throws IllegalPathException {
PartialPath sgName;
try {
sgName = new PartialPath(storageGroup);
} catch (IllegalPathException e) {
- logger.warn("Fail to get DeviceTimeIndexer for storage group {}, err:{}", storageGroup, e.getMessage());
- return null;
+ logger.warn("Fail to get TimeIndexer for storage group {}, err:{}", storageGroup, e.getMessage());
+ throw e;
}
lock.readLock().lock();
try {
@@ -118,7 +129,7 @@ public class IndexerManager {
}
}
- public DeviceTimeIndexer getUnseqIndexer(PartialPath storageGroup) {
+ public TimeIndexer getUnseqIndexer(PartialPath storageGroup) {
lock.readLock().lock();
try {
return unseqIndexers.get(storageGroup);
@@ -127,13 +138,13 @@ public class IndexerManager {
}
}
- public DeviceTimeIndexer getUnseqIndexer(String storageGroup) {
+ public TimeIndexer getUnseqIndexer(String storageGroup) throws IllegalPathException {
PartialPath sgName;
try {
sgName = new PartialPath(storageGroup);
} catch (IllegalPathException e) {
- logger.warn("Fail to get DeviceTimeIndexer for storage group {}, err:{}", storageGroup, e.getMessage());
- return null;
+ logger.warn("Fail to get TimeIndexer for storage group {}, err:{}", storageGroup, e.getMessage());
+ throw e;
}
lock.readLock().lock();
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceIndex.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndex.java
similarity index 80%
rename from server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceIndex.java
rename to server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndex.java
index bafa0c1..261033d 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceIndex.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndex.java
@@ -16,25 +16,25 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.timeIndex.device;
+package org.apache.iotdb.db.timeIndex;
import org.apache.iotdb.db.metadata.PartialPath;
/**
- * Device Index, like [(deviceId, startTime, endTime, TsFilePath)] to accelerate query
+ * Time Index, like deviceIndexer: [(deviceId, startTime, endTime, TsFilePath)] to accelerate query
*/
-public class DeviceIndex {
- private PartialPath[] deviceIds;
+public class TimeIndex {
+ private PartialPath[] paths;
private long[] startTimes;
private long[] endTimes;
private String tsFilePath;
- public PartialPath[] getDeviceIds() {
- return deviceIds;
+ public PartialPath[] getPaths() {
+ return paths;
}
- public void setDeviceIds(PartialPath[] deviceIds) {
- this.deviceIds = deviceIds;
+ public void setPaths(PartialPath[] paths) {
+ this.paths = paths;
}
public long[] getStartTimes() {
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexer.java
index a2d9d17..36e9297 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexer.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexer.java
@@ -1,8 +1,25 @@
+/*
+ * 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.timeIndex;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.PartialPath;
-import org.apache.iotdb.db.timeIndex.device.UpdateIndexsParam;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.util.List;
@@ -55,7 +72,7 @@ public interface TimeIndexer {
* @param tsFilePath
* @return
*/
- public boolean addIndexForPaths(Map<PartialPath, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath);
+ public boolean addIndexForPaths(Map<String, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath);
/**
* delete one index for the path
@@ -85,14 +102,14 @@ public interface TimeIndexer {
* @param tsFilePath
* @return
*/
- public boolean deleteIndexForDevices(Map<PartialPath, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath);
+ public boolean deleteIndexForPaths(Map<String, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath);
/**
* after merge, we should keep the index is updated in consistency
* @param updateIndexsParam
* @return whether success
*/
- public boolean updateIndexForDevices(UpdateIndexsParam updateIndexsParam);
+ public boolean updateIndexForPaths(UpdateIndexsParam updateIndexsParam);
/**
* found the related tsFile(only cover sealed tsfile) for one deviceId
@@ -100,5 +117,5 @@ public interface TimeIndexer {
* @param timeFilter
* @return whether success
*/
- public List<TsFileResource> filterByOneDevice(PartialPath path, Filter timeFilter);
+ public List<TsFileResource> filterByPath(PartialPath path, Filter timeFilter);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/UpdateIndexsParam.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java
similarity index 74%
rename from server/src/main/java/org/apache/iotdb/db/timeIndex/device/UpdateIndexsParam.java
rename to server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java
index e99fc3f..e7b14b7 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/UpdateIndexsParam.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java
@@ -16,28 +16,28 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.timeIndex.device;
+package org.apache.iotdb.db.timeIndex;
/**
- * to atomic update indexs for device
+ * to atomic update indexs for path
*/
public class UpdateIndexsParam {
- private DeviceIndex[] oldIndexs;
- private DeviceIndex newIndex;
+ private TimeIndex[] oldIndexs;
+ private TimeIndex newIndex;
- public DeviceIndex[] getOldIndexs() {
+ public TimeIndex[] getOldIndexs() {
return oldIndexs;
}
- public void setOldIndexs(DeviceIndex[] oldIndexs) {
+ public void setOldIndexs(TimeIndex[] oldIndexs) {
this.oldIndexs = oldIndexs;
}
- public DeviceIndex getNewIndex() {
+ public TimeIndex getNewIndex() {
return newIndex;
}
- public void setNewIndex(DeviceIndex newIndex) {
+ public void setNewIndex(TimeIndex newIndex) {
this.newIndex = newIndex;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceTimeIndexer.java
deleted file mode 100644
index d846d0d..0000000
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceTimeIndexer.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.db.timeIndex.device;
-
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.metadata.PartialPath;
-import org.apache.iotdb.tsfile.read.filter.basic.Filter;
-
-import java.util.List;
-import java.util.Map;
-
-/**
- * manage the index [deviceId, (startTime, endTime, TsFilePath)] for all devices
- */
-public interface DeviceTimeIndexer {
- /**
- * init the Indexer when IoTDB start
- * @return whether success
- */
- public boolean init();
-
- /**
- * may do some prepared work before operation
- * @return whether success
- */
- public boolean begin();
-
- /**
- * may do some resource release after operation
- * @return whether success
- */
- public boolean end();
-
- /**
- * add one index for the deviceId
- * @param deviceId
- * @param startTime
- * @param endTime
- * @param tsFilePath
- * @return whether success
- */
- public boolean addIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath);
-
- /**
- * add one index for the deviceId
- * @param deviceId
- * @param startTime
- * @param endTime
- * @param tsFilePath
- * @return
- */
- public boolean addIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath);
-
- /**
- * add all indexs for a flushed tsFile
- * @param deviceIds
- * @param startTimes
- * @param endTimes
- * @param tsFilePath
- * @return
- */
- public boolean addIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath);
-
- /**
- * delete one index for the deviceId
- * @param deviceId
- * @param startTime
- * @param endTime
- * @param tsFilePath
- * @return whether success
- */
- public boolean deleteIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath);
-
- /**
- * delete one index for the deviceId
- * @param deviceId
- * @param startTime
- * @param endTime
- * @param tsFilePath
- * @return
- */
- public boolean deleteIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath);
-
- /**
- * delete all index for one deleted tsfile
- * @param deviceIds
- * @param startTimes
- * @param endTimes
- * @param tsFilePath
- * @return
- */
- public boolean deleteIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath);
-
- /**
- * after merge, we should keep the index is updated in consistency
- * @param updateIndexsParam
- * @return whether success
- */
- public boolean updateIndexForDevices(UpdateIndexsParam updateIndexsParam);
-
- /**
- * found the related tsFile(only cover sealed tsfile) for one deviceId
- * @param deviceId
- * @param timeFilter
- * @return whether success
- */
- public List<TsFileResource> filterByOneDevice(PartialPath deviceId, Filter timeFilter);
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllTimeIndexer.java
index c5dab56..7cd51f7 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllTimeIndexer.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllTimeIndexer.java
@@ -20,12 +20,14 @@ package org.apache.iotdb.db.timeIndex.device;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.timeIndex.TimeIndexer;
+import org.apache.iotdb.db.timeIndex.UpdateIndexsParam;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.util.List;
import java.util.Map;
-public class LoadAllTimeIndexer implements DeviceTimeIndexer {
+public class LoadAllTimeIndexer implements TimeIndexer {
public LoadAllTimeIndexer() {
@@ -47,42 +49,42 @@ public class LoadAllTimeIndexer implements DeviceTimeIndexer {
}
@Override
- public boolean addIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath) {
+ public boolean addIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath) {
return false;
}
@Override
- public boolean addIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath) {
+ public boolean addIndexForPath(String path, long startTime, long endTime, String tsFilePath) {
return false;
}
@Override
- public boolean addIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath) {
+ public boolean addIndexForPaths(Map<String, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath) {
return false;
}
@Override
- public boolean deleteIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath) {
+ public boolean deleteIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath) {
return false;
}
@Override
- public boolean deleteIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath) {
+ public boolean deleteIndexForPath(String path, long startTime, long endTime, String tsFilePath) {
return false;
}
@Override
- public boolean deleteIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath) {
+ public boolean deleteIndexForPaths(Map<String, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath) {
return false;
}
@Override
- public boolean updateIndexForDevices(UpdateIndexsParam updateIndexsParam) {
+ public boolean updateIndexForPaths(UpdateIndexsParam updateIndexsParam) {
return false;
}
@Override
- public List<TsFileResource> filterByOneDevice(PartialPath deviceId, Filter timeFilter) {
+ public List<TsFileResource> filterByPath(PartialPath path, Filter timeFilter) {
return null;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
index 62fa0cf..9635232 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
@@ -20,12 +20,14 @@ package org.apache.iotdb.db.timeIndex.device;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.timeIndex.TimeIndexer;
+import org.apache.iotdb.db.timeIndex.UpdateIndexsParam;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.util.List;
import java.util.Map;
-public class RocksDBDeviceTimeIndexer implements DeviceTimeIndexer {
+public class RocksDBDeviceTimeIndexer implements TimeIndexer {
@Override
public boolean init() {
return false;
@@ -42,42 +44,42 @@ public class RocksDBDeviceTimeIndexer implements DeviceTimeIndexer {
}
@Override
- public boolean addIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath) {
+ public boolean addIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath) {
return false;
}
@Override
- public boolean addIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath) {
+ public boolean addIndexForPath(String path, long startTime, long endTime, String tsFilePath) {
return false;
}
@Override
- public boolean addIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath) {
+ public boolean addIndexForPaths(Map<String, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath) {
return false;
}
@Override
- public boolean deleteIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath) {
+ public boolean deleteIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath) {
return false;
}
@Override
- public boolean deleteIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath) {
+ public boolean deleteIndexForPath(String path, long startTime, long endTime, String tsFilePath) {
return false;
}
@Override
- public boolean deleteIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath) {
+ public boolean deleteIndexForPaths(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath) {
return false;
}
@Override
- public boolean updateIndexForDevices(UpdateIndexsParam updateIndexsParam) {
+ public boolean updateIndexForPaths(UpdateIndexsParam updateIndexsParam) {
return false;
}
@Override
- public List<TsFileResource> filterByOneDevice(PartialPath deviceId, Filter timeFilter) {
+ public List<TsFileResource> filterByPath(PartialPath path, Filter timeFilter) {
return null;
}
}
[iotdb] 01/05: [IOTDB-698] add time indexer interface
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chaow pushed a commit to branch fileIndex
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3077da8501f91b5b06b06ca1e97c9c21b5be7c6a
Author: mychaow <94...@qq.com>
AuthorDate: Tue Aug 25 20:03:23 2020 +0800
[IOTDB-698] add time indexer interface
---
server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceIndex.java | 4 ++++
.../main/java/org/apache/iotdb/db/timeIndex/DeviceTimeIndexer.java | 4 ++++
.../src/main/java/org/apache/iotdb/db/timeIndex/IndexConstants.java | 4 ++++
.../src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java | 4 ++++
.../main/java/org/apache/iotdb/db/timeIndex/LoadAllTimeIndexer.java | 4 ++++
.../java/org/apache/iotdb/db/timeIndex/RocksDBDeviceTimeIndexer.java | 4 ++++
.../main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java | 4 ++++
7 files changed, 28 insertions(+)
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceIndex.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceIndex.java
new file mode 100644
index 0000000..8e37f8d
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceIndex.java
@@ -0,0 +1,4 @@
+package org.apache.iotdb.db.timeIndex;
+
+public class DeviceIndex {
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceTimeIndexer.java
new file mode 100644
index 0000000..8ca9dc2
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceTimeIndexer.java
@@ -0,0 +1,4 @@
+package org.apache.iotdb.db.timeIndex;
+
+public interface TimeIndexer {
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexConstants.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexConstants.java
new file mode 100644
index 0000000..296bb57
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexConstants.java
@@ -0,0 +1,4 @@
+package org.apache.iotdb.db.timeIndex;
+
+public class IndexConstants {
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java
new file mode 100644
index 0000000..4ce6e38
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java
@@ -0,0 +1,4 @@
+package org.apache.iotdb.db.timeIndex;
+
+public class IndexerManager {
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/LoadAllTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/LoadAllTimeIndexer.java
new file mode 100644
index 0000000..c553213
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/LoadAllTimeIndexer.java
@@ -0,0 +1,4 @@
+package org.apache.iotdb.db.timeIndex;
+
+public class LoadAllTimeIndexer {
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/RocksDBDeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/RocksDBDeviceTimeIndexer.java
new file mode 100644
index 0000000..219b7f0
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/RocksDBDeviceTimeIndexer.java
@@ -0,0 +1,4 @@
+package org.apache.iotdb.db.timeIndex;
+
+public class RocksDBDeviceTimeIndexer {
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java
new file mode 100644
index 0000000..80a7845
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java
@@ -0,0 +1,4 @@
+package org.apache.iotdb.db.timeIndex;
+
+public class UpdateIndexParam {
+}
[iotdb] 04/05: rename IndexerManager to FileIndexerManager
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chaow pushed a commit to branch fileIndex
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ba2ecdac2c3a6c772bcb4eaa77a26558b662c4b2
Author: chaow <xu...@gmail.com>
AuthorDate: Wed Sep 16 14:29:53 2020 +0800
rename IndexerManager to FileIndexerManager
---
.../resources/conf/iotdb-engine.properties | 11 ++--
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 20 +++----
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 9 +--
.../org/apache/iotdb/db/engine/StorageEngine.java | 14 ++---
.../iotdb/db/engine/merge/task/MergeFileTask.java | 10 ++--
.../engine/storagegroup/StorageGroupProcessor.java | 70 +++++++++++-----------
.../db/engine/storagegroup/TsFileProcessor.java | 17 +++---
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 2 -
...IndexerManager.java => FileIndexerManager.java} | 30 +++++-----
.../{TimeIndexer.java => FileTimeIndexer.java} | 2 +-
.../org/apache/iotdb/db/timeIndex/TimeIndex.java | 2 +-
...eIndexer.java => LoadAllDeviceTimeIndexer.java} | 6 +-
.../timeIndex/device/RocksDBDeviceTimeIndexer.java | 4 +-
13 files changed, 95 insertions(+), 102 deletions(-)
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties b/server/src/assembly/resources/conf/iotdb-engine.properties
index 4b33bfc..f8b2ca4 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -229,7 +229,6 @@ write_read_schema_free_memory_proportion=4:3:1:2
# primitive array size (length of each array) in array pool
primitive_array_size=128
-<<<<<<< HEAD
# Ratio of write memory for invoking flush disk, 0.4 by default
flush_proportion=0.4
@@ -254,12 +253,10 @@ max_waiting_time_when_insert_blocked=0
# estimated metadata size (in byte) of one timeseries in Mtree
estimated_series_size=300
-=======
-# device time indexer
-enable_device_indexer=false
-# 0, indexer base on file; 1, indexer base on rocksdb
-device_indexer_type=0
->>>>>>> fix conflict
+
+enable_file_time_indexer=false
+# 0, device indexer base on file; 1, device indexer base on rocksdb
+file_time_indexer_type=0
####################
### Upgrade Configurations
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index c571022..52a1c63 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -783,8 +783,8 @@ public class IoTDBConfig {
private boolean debugState = false;
// enable new indexer
- private boolean enableDeviceIndexer = false;
- private int deviceIndexerType = 0;
+ private boolean enableFileTimeIndexer = false;
+ private int fileTimeIndexerType = 0;
public IoTDBConfig() {
// empty constructor
@@ -2098,19 +2098,19 @@ public class IoTDBConfig {
this.defaultIndexWindowRange = defaultIndexWindowRange;
}
- public boolean isEnableDeviceIndexer() {
- return enableDeviceIndexer;
+ public boolean isEnableFileTimeIndexer() {
+ return enableFileTimeIndexer;
}
- public void setEnableDeviceIndexer(boolean enableDeviceIndexer) {
- this.enableDeviceIndexer = enableDeviceIndexer;
+ public void setEnableFileTimeIndexer(boolean enableFileTimeIndexer) {
+ this.enableFileTimeIndexer = enableFileTimeIndexer;
}
- public int getDeviceIndexerType() {
- return deviceIndexerType;
+ public int getFileTimeIndexerType() {
+ return fileTimeIndexerType;
}
- public void setDeviceIndexerType(int deviceIndexerType) {
- this.deviceIndexerType = deviceIndexerType;
+ public void setFileTimeIndexerType(int fileTimeIndexerType) {
+ this.fileTimeIndexerType = fileTimeIndexerType;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 8b559db..33d8e8a 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -517,13 +517,14 @@ public class IoTDBDescriptor {
conf.setDebugState(Boolean.parseBoolean(properties
.getProperty("debug_state", String.valueOf(conf.isDebugOn()))));
- conf.setEnableDeviceIndexer((Boolean.parseBoolean(
+
+ conf.setEnableFileTimeIndexer((Boolean.parseBoolean(
properties.getProperty(
- "enable_device_indexer", String.valueOf(conf.isEnableDeviceIndexer())))));
+ "enable_file_time_indexer", String.valueOf(conf.isEnableFileTimeIndexer())))));
- conf.setDeviceIndexerType((Integer.parseInt(
+ conf.setFileTimeIndexerType((Integer.parseInt(
properties.getProperty(
- "device_indexer_type", String.valueOf(conf.getDeviceIndexerType())))));
+ "file_time_indexer_type", String.valueOf(conf.getFileTimeIndexerType())))));
// mqtt
if (properties.getProperty(IoTDBConstant.MQTT_HOST_NAME) != null) {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
index d785627..b17f413 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
@@ -78,7 +78,7 @@ import org.apache.iotdb.db.query.control.QueryFileManager;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.service.ServiceType;
-import org.apache.iotdb.db.timeIndex.IndexerManager;
+import org.apache.iotdb.db.timeIndex.FileIndexerManager;
import org.apache.iotdb.db.utils.FilePathUtils;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.db.utils.UpgradeUtils;
@@ -628,9 +628,9 @@ public class StorageEngine implements IService {
logger.info("Start deleting all storage groups' timeseries");
syncCloseAllProcessor();
for (PartialPath storageGroup : IoTDB.metaManager.getAllStorageGroupPaths()) {
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- IndexerManager.getInstance().deleteSeqIndexer(storageGroup);
- IndexerManager.getInstance().deleteUnseqIndexer(storageGroup);
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
+ FileIndexerManager.getInstance().deleteSeqIndexer(storageGroup);
+ FileIndexerManager.getInstance().deleteUnseqIndexer(storageGroup);
}
this.deleteAllDataFilesInOneStorageGroup(storageGroup);
}
@@ -645,9 +645,9 @@ public class StorageEngine implements IService {
public void deleteStorageGroup(PartialPath storageGroupPath) {
deleteAllDataFilesInOneStorageGroup(storageGroupPath);
StorageGroupProcessor processor = processorMap.remove(storageGroupPath);
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- IndexerManager.getInstance().deleteSeqIndexer(storageGroupPath);
- IndexerManager.getInstance().deleteUnseqIndexer(storageGroupPath);
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
+ FileIndexerManager.getInstance().deleteSeqIndexer(storageGroupPath);
+ FileIndexerManager.getInstance().deleteUnseqIndexer(storageGroupPath);
}
if (processor != null) {
processor.deleteFolder(systemDir);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
index ac645d6..a806271 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
@@ -41,8 +41,8 @@ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.query.control.FileReaderManager;
-import org.apache.iotdb.db.timeIndex.IndexerManager;
-import org.apache.iotdb.db.timeIndex.TimeIndexer;
+import org.apache.iotdb.db.timeIndex.FileIndexerManager;
+import org.apache.iotdb.db.timeIndex.FileTimeIndexer;
import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
@@ -202,11 +202,11 @@ class MergeFileTask {
fsFactory.moveFile(fileWriter.getFile(), newMergeFile);
seqFile.setFile(newMergeFile);
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
// add new device index
// may Indexer need delete old index background
- TimeIndexer timeIndexer = IndexerManager.getInstance().getSeqIndexer(seqFile.getStorageGroupName());
- timeIndexer.addIndexForPaths(seqFile.getDeviceToIndexMap(), seqFile.getStartTimes(), seqFile.getEndTimes(), seqFile.getTsFilePath());
+ FileTimeIndexer fileTimeIndexer = FileIndexerManager.getInstance().getSeqIndexer(seqFile.getStorageGroupName());
+ fileTimeIndexer.addIndexForPaths(seqFile.getDeviceToIndexMap(), seqFile.getStartTimes(), seqFile.getEndTimes(), seqFile.getTsFilePath());
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index 11dcf16..2e03646 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -82,8 +82,8 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryFileManager;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.service.UpgradeSevice;
-import org.apache.iotdb.db.timeIndex.IndexerManager;
-import org.apache.iotdb.db.timeIndex.TimeIndexer;
+import org.apache.iotdb.db.timeIndex.FileIndexerManager;
+import org.apache.iotdb.db.timeIndex.FileTimeIndexer;
import org.apache.iotdb.db.utils.CopyOnReadLinkedList;
import org.apache.iotdb.db.writelog.recover.TsFileRecoverPerformer;
import org.apache.iotdb.rpc.RpcUtils;
@@ -270,8 +270,8 @@ public class StorageGroupProcessor {
* unseqDeviceTimeIndexer manage the device Index of unseq tsfiles
*
*/
- private TimeIndexer seqTimeIndexer;
- private TimeIndexer unseqTimeIndexer;
+ private FileTimeIndexer seqFileTimeIndexer;
+ private FileTimeIndexer unseqFileTimeIndexer;
public StorageGroupProcessor(String systemDir, String storageGroupName,
TsFileFlushPolicy fileFlushPolicy) throws StorageGroupProcessorException {
@@ -385,14 +385,14 @@ public class StorageGroupProcessor {
globalLatestFlushedTimeForEachDevice.putAll(endTimeMap);
}
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
// init TimeIndexer
try {
- seqTimeIndexer = IndexerManager.getInstance().getSeqIndexer(storageGroupName);
- unseqTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(storageGroupName);
- if (seqTimeIndexer != null) {
- seqTimeIndexer.init();
- unseqTimeIndexer.init();
+ seqFileTimeIndexer = FileIndexerManager.getInstance().getSeqIndexer(storageGroupName);
+ unseqFileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(storageGroupName);
+ if (seqFileTimeIndexer != null) {
+ seqFileTimeIndexer.init();
+ unseqFileTimeIndexer.init();
}
} catch (IllegalPathException e) {
throw new StorageGroupProcessorException(e);
@@ -1303,15 +1303,15 @@ public class StorageGroupProcessor {
}
tsFileManagement.remove(resource, isSeq);
if (isSeq) {
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- TimeIndexer timeIndexer = IndexerManager.getInstance().getSeqIndexer(resource.getStorageGroupName());
- timeIndexer.deleteIndexForPaths(resource.getDeviceToIndexMap(), resource.getStartTimes(),
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
+ FileTimeIndexer fileTimeIndexer = FileIndexerManager.getInstance().getSeqIndexer(resource.getStorageGroupName());
+ fileTimeIndexer.deleteIndexForPaths(resource.getDeviceToIndexMap(), resource.getStartTimes(),
resource.getEndTimes(), resource.getTsFilePath());
}
} else {
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- TimeIndexer timeIndexer = IndexerManager.getInstance().getUnseqIndexer(resource.getStorageGroupName());
- timeIndexer.deleteIndexForPaths(resource.getDeviceToIndexMap(), resource.getStartTimes(),
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
+ FileTimeIndexer fileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(resource.getStorageGroupName());
+ fileTimeIndexer.deleteIndexForPaths(resource.getDeviceToIndexMap(), resource.getStartTimes(),
resource.getEndTimes(), resource.getTsFilePath());
}
}
@@ -1403,9 +1403,9 @@ public class StorageGroupProcessor {
List<TsFileResource> seqResources;
List<TsFileResource> unseqResources;
try {
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- seqResources = seqTimeIndexer.filterByPath(deviceId, timeFilter);
- unseqResources = unseqTimeIndexer.filterByPath(deviceId, timeFilter);
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
+ seqResources = seqFileTimeIndexer.filterByPath(deviceId, timeFilter);
+ unseqResources = unseqFileTimeIndexer.filterByPath(deviceId, timeFilter);
List<TsFileResource> unsealedSeqFiles = getUnSealedListResourceForQuery(
tsFileManagement.getTsFileList(true),
deviceId, measurementId, context, timeFilter, true);
@@ -1895,14 +1895,14 @@ public class StorageGroupProcessor {
newFilePartitionId)) {
updateLatestTimeMap(newTsFileResource);
}
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
- TimeIndexer timeIndexer = null;
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
+ FileTimeIndexer fileTimeIndexer = null;
if (newTsFileResource.isSeq()) {
- timeIndexer = IndexerManager.getInstance().getSeqIndexer(newTsFileResource.getStorageGroupName());
+ fileTimeIndexer = FileIndexerManager.getInstance().getSeqIndexer(newTsFileResource.getStorageGroupName());
} else {
- timeIndexer = IndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
+ fileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
}
- timeIndexer.addIndexForPaths(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
+ fileTimeIndexer.addIndexForPaths(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
newTsFileResource.getEndTimes(), newTsFileResource.getTsFilePath());
}
} catch (DiskSpaceInsufficientException | IllegalPathException e) {
@@ -1965,15 +1965,15 @@ public class StorageGroupProcessor {
newFilePartitionId);
}
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
try {
- TimeIndexer timeIndexer = null;
+ FileTimeIndexer fileTimeIndexer = null;
if (newTsFileResource.isSeq()) {
- timeIndexer = IndexerManager.getInstance().getSeqIndexer(newTsFileResource.getStorageGroupName());
+ fileTimeIndexer = FileIndexerManager.getInstance().getSeqIndexer(newTsFileResource.getStorageGroupName());
} else {
- timeIndexer = IndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
+ fileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
}
- timeIndexer.addIndexForPaths(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
+ fileTimeIndexer.addIndexForPaths(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
newTsFileResource.getEndTimes(), newTsFileResource.getTsFilePath());
} catch (IllegalPathException e) {
logger.error("Fail to get DeviceTimeIndexer for storage group {}, err:{}", newTsFileResource.getStorageGroupName(), e.getMessage());
@@ -2382,16 +2382,16 @@ public class StorageGroupProcessor {
tsFileResourceToBeDeleted.writeLock();
try {
tsFileResourceToBeDeleted.remove();
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
try {
- TimeIndexer timeIndexer = null;
+ FileTimeIndexer fileTimeIndexer = null;
if (tsFileResourceToBeDeleted.isSeq()) {
- timeIndexer = IndexerManager.getInstance().getSeqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
+ fileTimeIndexer = FileIndexerManager.getInstance().getSeqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
} else {
- timeIndexer = IndexerManager.getInstance().getUnseqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
+ fileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
}
- if (timeIndexer != null) {
- timeIndexer.deleteIndexForPaths(tsFileResourceToBeDeleted.getDeviceToIndexMap(),
+ if (fileTimeIndexer != null) {
+ fileTimeIndexer.deleteIndexForPaths(tsFileResourceToBeDeleted.getDeviceToIndexMap(),
tsFileResourceToBeDeleted.getStartTimes(), tsFileResourceToBeDeleted.getEndTimes(),
tsFileResourceToBeDeleted.getTsFilePath());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index d44c281..cf18c83 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -50,11 +50,8 @@ import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.UpdateEndTi
import org.apache.iotdb.db.engine.version.VersionController;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.WriteProcessException;
-<<<<<<< HEAD
import org.apache.iotdb.db.exception.metadata.MetadataException;
-=======
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
->>>>>>> fix conflict
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
@@ -63,8 +60,8 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.rescon.PrimitiveArrayManager;
import org.apache.iotdb.db.rescon.SystemInfo;
import org.apache.iotdb.db.utils.MemUtils;
-import org.apache.iotdb.db.timeIndex.IndexerManager;
-import org.apache.iotdb.db.timeIndex.TimeIndexer;
+import org.apache.iotdb.db.timeIndex.FileIndexerManager;
+import org.apache.iotdb.db.timeIndex.FileTimeIndexer;
import org.apache.iotdb.db.utils.QueryUtils;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.iotdb.db.writelog.WALFlushListener;
@@ -818,16 +815,16 @@ public class TsFileProcessor {
private void endFile() throws IOException, TsFileProcessorException {
long closeStartTime = System.currentTimeMillis();
- if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
// update device index
try {
- TimeIndexer timeIndexer = null;
+ FileTimeIndexer fileTimeIndexer = null;
if (sequence) {
- timeIndexer = IndexerManager.getInstance().getSeqIndexer(storageGroupName);
+ fileTimeIndexer = FileIndexerManager.getInstance().getSeqIndexer(storageGroupName);
} else {
- timeIndexer = IndexerManager.getInstance().getUnseqIndexer(storageGroupName);
+ fileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(storageGroupName);
}
- timeIndexer.addIndexForPaths(tsFileResource.deviceToIndex, tsFileResource.startTimes,
+ fileTimeIndexer.addIndexForPaths(tsFileResource.deviceToIndex, tsFileResource.startTimes,
tsFileResource.endTimes, tsFileResource.getTsFilePath());
} catch (IllegalPathException e) {
logger.error("Failed to endFile {} for storage group {}, err:{}", tsFileResource.getTsFile(),
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index ccf31b0..ea828c2 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -127,7 +127,6 @@ import org.apache.iotdb.db.query.dataset.SingleDataSet;
import org.apache.iotdb.db.query.executor.IQueryRouter;
import org.apache.iotdb.db.query.executor.QueryRouter;
import org.apache.iotdb.db.service.IoTDB;
-import org.apache.iotdb.db.timeIndex.IndexerManager;
import org.apache.iotdb.db.utils.AuthUtils;
import org.apache.iotdb.db.utils.FileLoaderUtils;
import org.apache.iotdb.db.utils.UpgradeUtils;
@@ -140,7 +139,6 @@ import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.common.RowRecord;
-import org.apache.iotdb.tsfile.read.filter.operator.In;
import org.apache.iotdb.tsfile.read.query.dataset.EmptyDataSet;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.iotdb.tsfile.utils.Binary;
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/FileIndexerManager.java
similarity index 78%
rename from server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java
rename to server/src/main/java/org/apache/iotdb/db/timeIndex/FileIndexerManager.java
index eb09ab8..2c9f602 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/FileIndexerManager.java
@@ -29,12 +29,12 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-public class IndexerManager {
+public class FileIndexerManager {
private String indexerFilePath;
- private Map<PartialPath, TimeIndexer> seqIndexers;
- private Map<PartialPath, TimeIndexer> unseqIndexers;
+ private Map<PartialPath, FileTimeIndexer> seqIndexers;
+ private Map<PartialPath, FileTimeIndexer> unseqIndexers;
private ReentrantReadWriteLock lock;
- private static final Logger logger = LoggerFactory.getLogger(IndexerManager.class);
+ private static final Logger logger = LoggerFactory.getLogger(FileIndexerManager.class);
private static class IndexerManagerHolder {
@@ -42,14 +42,14 @@ public class IndexerManager {
// allowed to do nothing
}
- private static final IndexerManager INSTANCE = new IndexerManager();
+ private static final FileIndexerManager INSTANCE = new FileIndexerManager();
}
- public static IndexerManager getInstance() {
+ public static FileIndexerManager getInstance() {
return IndexerManagerHolder.INSTANCE;
}
- private IndexerManager() {
+ private FileIndexerManager() {
indexerFilePath = IoTDBDescriptor.getInstance().getConfig().getSchemaDir()
+ File.pathSeparator + IndexConstants.INDEXER_FILE;
seqIndexers = new ConcurrentHashMap<>();
@@ -68,19 +68,19 @@ public class IndexerManager {
return true;
}
- public void addSeqIndexer(PartialPath storageGroup, TimeIndexer TimeIndexer) {
+ public void addSeqIndexer(PartialPath storageGroup, FileTimeIndexer fileTimeIndexer) {
lock.writeLock().lock();
try {
- seqIndexers.put(storageGroup, TimeIndexer);
+ seqIndexers.put(storageGroup, fileTimeIndexer);
} finally {
lock.writeLock().unlock();
}
}
- public void addUnseqIndexer(PartialPath storageGroup, TimeIndexer TimeIndexer) {
+ public void addUnseqIndexer(PartialPath storageGroup, FileTimeIndexer fileTimeIndexer) {
lock.writeLock().lock();
try {
- unseqIndexers.put(storageGroup, TimeIndexer);
+ unseqIndexers.put(storageGroup, fileTimeIndexer);
} finally {
lock.writeLock().unlock();
}
@@ -104,7 +104,7 @@ public class IndexerManager {
}
}
- public TimeIndexer getSeqIndexer(PartialPath storageGroup) {
+ public FileTimeIndexer getSeqIndexer(PartialPath storageGroup) {
lock.readLock().lock();
try {
return seqIndexers.get(storageGroup);
@@ -113,7 +113,7 @@ public class IndexerManager {
}
}
- public TimeIndexer getSeqIndexer(String storageGroup) throws IllegalPathException {
+ public FileTimeIndexer getSeqIndexer(String storageGroup) throws IllegalPathException {
PartialPath sgName;
try {
sgName = new PartialPath(storageGroup);
@@ -129,7 +129,7 @@ public class IndexerManager {
}
}
- public TimeIndexer getUnseqIndexer(PartialPath storageGroup) {
+ public FileTimeIndexer getUnseqIndexer(PartialPath storageGroup) {
lock.readLock().lock();
try {
return unseqIndexers.get(storageGroup);
@@ -138,7 +138,7 @@ public class IndexerManager {
}
}
- public TimeIndexer getUnseqIndexer(String storageGroup) throws IllegalPathException {
+ public FileTimeIndexer getUnseqIndexer(String storageGroup) throws IllegalPathException {
PartialPath sgName;
try {
sgName = new PartialPath(storageGroup);
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/FileTimeIndexer.java
similarity index 98%
rename from server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexer.java
rename to server/src/main/java/org/apache/iotdb/db/timeIndex/FileTimeIndexer.java
index 36e9297..64c6dab 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexer.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/FileTimeIndexer.java
@@ -25,7 +25,7 @@ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.util.List;
import java.util.Map;
-public interface TimeIndexer {
+public interface FileTimeIndexer {
/**
* init the Indexer when IoTDB start
* @return whether success
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndex.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndex.java
index 261033d..5cf3705 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndex.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndex.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.timeIndex;
import org.apache.iotdb.db.metadata.PartialPath;
/**
- * Time Index, like deviceIndexer: [(deviceId, startTime, endTime, TsFilePath)] to accelerate query
+ * Time Index, like deviceIndex: [(deviceId, startTime, endTime, TsFilePath)] to accelerate query
*/
public class TimeIndex {
private PartialPath[] paths;
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllDeviceTimeIndexer.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllTimeIndexer.java
rename to server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllDeviceTimeIndexer.java
index 7cd51f7..3d02b9c 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllTimeIndexer.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllDeviceTimeIndexer.java
@@ -20,16 +20,16 @@ package org.apache.iotdb.db.timeIndex.device;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.PartialPath;
-import org.apache.iotdb.db.timeIndex.TimeIndexer;
+import org.apache.iotdb.db.timeIndex.FileTimeIndexer;
import org.apache.iotdb.db.timeIndex.UpdateIndexsParam;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.util.List;
import java.util.Map;
-public class LoadAllTimeIndexer implements TimeIndexer {
+public class LoadAllDeviceTimeIndexer implements FileTimeIndexer {
- public LoadAllTimeIndexer() {
+ public LoadAllDeviceTimeIndexer() {
}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
index 9635232..686531f 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
@@ -20,14 +20,14 @@ package org.apache.iotdb.db.timeIndex.device;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.PartialPath;
-import org.apache.iotdb.db.timeIndex.TimeIndexer;
+import org.apache.iotdb.db.timeIndex.FileTimeIndexer;
import org.apache.iotdb.db.timeIndex.UpdateIndexsParam;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.util.List;
import java.util.Map;
-public class RocksDBDeviceTimeIndexer implements TimeIndexer {
+public class RocksDBDeviceTimeIndexer implements FileTimeIndexer {
@Override
public boolean init() {
return false;
[iotdb] 02/05: [IOTDB-698] add time indexer interface
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chaow pushed a commit to branch fileIndex
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f6c752dbb7316c113e578664710e0ff2e3447c59
Author: mychaow <94...@qq.com>
AuthorDate: Tue Aug 25 20:03:42 2020 +0800
[IOTDB-698] add time indexer interface
---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 20 +++
.../org/apache/iotdb/db/engine/StorageEngine.java | 9 ++
.../iotdb/db/engine/merge/task/MergeFileTask.java | 11 ++
.../engine/storagegroup/StorageGroupProcessor.java | 127 ++++++++++++++++++-
.../db/engine/storagegroup/TsFileProcessor.java | 12 ++
.../db/engine/storagegroup/TsFileResource.java | 19 ++-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 2 +
.../org/apache/iotdb/db/timeIndex/DeviceIndex.java | 4 -
.../iotdb/db/timeIndex/DeviceTimeIndexer.java | 4 -
.../apache/iotdb/db/timeIndex/IndexConstants.java | 42 ++++++
.../apache/iotdb/db/timeIndex/IndexerManager.java | 141 +++++++++++++++++++++
.../iotdb/db/timeIndex/LoadAllTimeIndexer.java | 4 -
.../db/timeIndex/RocksDBDeviceTimeIndexer.java | 4 -
.../org/apache/iotdb/db/timeIndex/TimeIndexer.java | 104 +++++++++++++++
.../iotdb/db/timeIndex/UpdateIndexsParam.java | 4 -
.../iotdb/db/timeIndex/device/DeviceIndex.java | 63 +++++++++
.../db/timeIndex/device/DeviceTimeIndexer.java | 124 ++++++++++++++++++
.../db/timeIndex/device/LoadAllTimeIndexer.java | 88 +++++++++++++
.../timeIndex/device/RocksDBDeviceTimeIndexer.java | 83 ++++++++++++
.../db/timeIndex/device/UpdateIndexsParam.java | 43 +++++++
.../iotdb/tsfile/utils/ReadWriteIOUtils.java | 26 ++--
21 files changed, 888 insertions(+), 46 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index a1a922b..c571022 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -782,6 +782,10 @@ public class IoTDBConfig {
*/
private boolean debugState = false;
+ // enable new indexer
+ private boolean enableDeviceIndexer = false;
+ private int deviceIndexerType = 0;
+
public IoTDBConfig() {
// empty constructor
}
@@ -2093,4 +2097,20 @@ public class IoTDBConfig {
public void setDefaultIndexWindowRange(int defaultIndexWindowRange) {
this.defaultIndexWindowRange = defaultIndexWindowRange;
}
+
+ public boolean isEnableDeviceIndexer() {
+ return enableDeviceIndexer;
+ }
+
+ public void setEnableDeviceIndexer(boolean enableDeviceIndexer) {
+ this.enableDeviceIndexer = enableDeviceIndexer;
+ }
+
+ public int getDeviceIndexerType() {
+ return deviceIndexerType;
+ }
+
+ public void setDeviceIndexerType(int deviceIndexerType) {
+ this.deviceIndexerType = deviceIndexerType;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
index 4fd8b57..d785627 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
@@ -78,6 +78,7 @@ import org.apache.iotdb.db.query.control.QueryFileManager;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.db.timeIndex.IndexerManager;
import org.apache.iotdb.db.utils.FilePathUtils;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.db.utils.UpgradeUtils;
@@ -627,6 +628,10 @@ public class StorageEngine implements IService {
logger.info("Start deleting all storage groups' timeseries");
syncCloseAllProcessor();
for (PartialPath storageGroup : IoTDB.metaManager.getAllStorageGroupPaths()) {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ IndexerManager.getInstance().deleteSeqIndexer(storageGroup);
+ IndexerManager.getInstance().deleteUnseqIndexer(storageGroup);
+ }
this.deleteAllDataFilesInOneStorageGroup(storageGroup);
}
return true;
@@ -640,6 +645,10 @@ public class StorageEngine implements IService {
public void deleteStorageGroup(PartialPath storageGroupPath) {
deleteAllDataFilesInOneStorageGroup(storageGroupPath);
StorageGroupProcessor processor = processorMap.remove(storageGroupPath);
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ IndexerManager.getInstance().deleteSeqIndexer(storageGroupPath);
+ IndexerManager.getInstance().deleteUnseqIndexer(storageGroupPath);
+ }
if (processor != null) {
processor.deleteFolder(systemDir);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
index f148655..42486ca 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
@@ -40,6 +40,10 @@ import org.apache.iotdb.db.engine.merge.recover.MergeLogger;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.query.control.FileReaderManager;
+import org.apache.iotdb.db.timeIndex.device.DeviceIndex;
+import org.apache.iotdb.db.timeIndex.device.DeviceTimeIndexer;
+import org.apache.iotdb.db.timeIndex.IndexerManager;
+import org.apache.iotdb.db.timeIndex.device.UpdateIndexsParam;
import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
@@ -198,6 +202,13 @@ class MergeFileTask {
newMergeFile.delete();
fsFactory.moveFile(fileWriter.getFile(), newMergeFile);
seqFile.setFile(newMergeFile);
+
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ // add new device index
+ // may Indexer need delete old index background
+ DeviceTimeIndexer deviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(seqFile.getStorageGroupName());
+ deviceTimeIndexer.addIndexForDevices(seqFile.getDeviceToIndexMap(), seqFile.getStartTimes(), seqFile.getEndTimes(), seqFile.getTsFilePath());
+ }
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index 888f98e..90b3769 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -68,6 +68,7 @@ import org.apache.iotdb.db.exception.LoadFileException;
import org.apache.iotdb.db.exception.StorageGroupProcessorException;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.WriteProcessException;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.OutOfTTLException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
@@ -81,6 +82,8 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryFileManager;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.service.UpgradeSevice;
+import org.apache.iotdb.db.timeIndex.IndexerManager;
+import org.apache.iotdb.db.timeIndex.device.DeviceTimeIndexer;
import org.apache.iotdb.db.utils.CopyOnReadLinkedList;
import org.apache.iotdb.db.writelog.recover.TsFileRecoverPerformer;
import org.apache.iotdb.rpc.RpcUtils;
@@ -243,6 +246,7 @@ public class StorageGroupProcessor {
private Map<Long, Long> partitionMaxFileVersions = new HashMap<>();
/**
+
* value of root.stats."root.sg".TOTAL_POINTS
*/
private long monitorSeriesValue;
@@ -261,6 +265,14 @@ public class StorageGroupProcessor {
private List<CloseFileListener> customCloseFileListeners = Collections.emptyList();
private List<FlushListener> customFlushListeners = Collections.emptyList();
+ /*
+ * seqDeviceTimeIndexer manage the device Index of seq tsfiles
+ * unseqDeviceTimeIndexer manage the device Index of unseq tsfiles
+ *
+ */
+ private DeviceTimeIndexer seqDeviceTimeIndexer;
+ private DeviceTimeIndexer unseqDeviceTimeIndexer;
+
public StorageGroupProcessor(String systemDir, String storageGroupName,
TsFileFlushPolicy fileFlushPolicy) throws StorageGroupProcessorException {
this.storageGroupName = storageGroupName;
@@ -373,6 +385,16 @@ public class StorageGroupProcessor {
globalLatestFlushedTimeForEachDevice.putAll(endTimeMap);
}
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ // init TimeIndexer
+ seqDeviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(storageGroupName);
+ unseqDeviceTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(storageGroupName);
+ if (seqDeviceTimeIndexer != null) {
+ seqDeviceTimeIndexer.init();
+ unseqDeviceTimeIndexer.init();
+ }
+ }
+
}
public long getMonitorSeriesValue() {
@@ -898,8 +920,7 @@ public class StorageGroupProcessor {
// try to update the latest time of the device of this tsRecord
if (latestTimeForEachDevice.get(timePartitionId)
- .getOrDefault(insertRowPlan.getDeviceId().getFullPath(), Long.MIN_VALUE) < insertRowPlan
- .getTime()) {
+ .getOrDefault(insertRowPlan.getDeviceId().getFullPath(), Long.MIN_VALUE) < insertRowPlan.getTime()) {
latestTimeForEachDevice.get(timePartitionId)
.put(insertRowPlan.getDeviceId().getFullPath(), insertRowPlan.getTime());
}
@@ -1277,6 +1298,23 @@ public class StorageGroupProcessor {
new Date(timeLowerBound), dataTTL);
}
tsFileManagement.remove(resource, isSeq);
+ if (isSeq) {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ DeviceTimeIndexer deviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(resource.getStorageGroupName());
+ if (deviceTimeIndexer != null) {
+ deviceTimeIndexer.deleteIndexForDevices(resource.getDeviceToIndexMap(), resource.getStartTimes(),
+ resource.getEndTimes(), resource.getTsFilePath());
+ }
+ }
+ } else {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ DeviceTimeIndexer deviceTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(resource.getStorageGroupName());
+ if (deviceTimeIndexer != null) {
+ deviceTimeIndexer.deleteIndexForDevices(resource.getDeviceToIndexMap(), resource.getStartTimes(),
+ resource.getEndTimes(), resource.getTsFilePath());
+ }
+ }
+ }
} finally {
resource.writeUnlock();
}
@@ -1360,15 +1398,28 @@ public class StorageGroupProcessor {
public QueryDataSource query(PartialPath deviceId, String measurementId, QueryContext context,
QueryFileManager filePathsManager, Filter timeFilter) throws QueryProcessException {
insertLock.readLock().lock();
+ List<TsFileResource> seqResources;
+ List<TsFileResource> unseqResources;
try {
- List<TsFileResource> seqResources = getFileResourceListForQuery(
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ seqResources = seqDeviceTimeIndexer.filterByOneDevice(deviceId, timeFilter);
+ unseqResources = unseqDeviceTimeIndexer.filterByOneDevice(deviceId, timeFilter);
+ List<TsFileResource> unsealedSeqFiles = getUnSealedListResourceForQuery(
tsFileManagement.getTsFileList(true),
- upgradeSeqFileList, deviceId, measurementId, context, timeFilter, true);
- List<TsFileResource> unseqResources = getFileResourceListForQuery(
+ deviceId, measurementId, context, timeFilter, true);
+ List<TsFileResource> unsealedUnseqFiles = getUnSealedListResourceForQuery(
tsFileManagement.getTsFileList(false),
+ deviceId, measurementId, context, timeFilter, false);
+ seqResources.addAll(unsealedSeqFiles);
+ unsealedSeqFiles.addAll(unsealedUnseqFiles);
+ } else {
+ seqResources = getFileResourceListForQuery(tsFileManagement.getTsFileList(true),
+ upgradeSeqFileList, deviceId, measurementId, context, timeFilter, true);
+ unseqResources = getFileResourceListForQuery(tsFileManagement.getTsFileList(false),
upgradeUnseqFileList, deviceId, measurementId, context, timeFilter, false);
+ }
QueryDataSource dataSource = new QueryDataSource(deviceId,
- seqResources, unseqResources);
+ seqResources, unseqResources);
// used files should be added before mergeLock is unlocked, or they may be deleted by
// running merge
// is null only in tests
@@ -1842,6 +1893,16 @@ public class StorageGroupProcessor {
newFilePartitionId)) {
updateLatestTimeMap(newTsFileResource);
}
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ DeviceTimeIndexer deviceTimeIndexer = null;
+ if (newTsFileResource.isSeq()) {
+ deviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(newTsFileResource.getStorageGroupName());
+ } else {
+ deviceTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
+ }
+ deviceTimeIndexer.addIndexForDevices(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
+ newTsFileResource.getEndTimes(), newTsFileResource.getTsFilePath());
+ }
} catch (DiskSpaceInsufficientException e) {
logger.error(
"Failed to append the tsfile {} to storage group processor {} because the disk space is insufficient.",
@@ -1902,6 +1963,17 @@ public class StorageGroupProcessor {
newFilePartitionId);
}
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ DeviceTimeIndexer deviceTimeIndexer = null;
+ if (newTsFileResource.isSeq()) {
+ deviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(newTsFileResource.getStorageGroupName());
+ } else {
+ deviceTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
+ }
+ deviceTimeIndexer.addIndexForDevices(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
+ newTsFileResource.getEndTimes(), newTsFileResource.getTsFilePath());
+ }
+
// update latest time map
updateLatestTimeMap(newTsFileResource);
long partitionNum = newTsFileResource.getTimePartition();
@@ -2302,6 +2374,19 @@ public class StorageGroupProcessor {
tsFileResourceToBeDeleted.writeLock();
try {
tsFileResourceToBeDeleted.remove();
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ DeviceTimeIndexer deviceTimeIndexer = null;
+ if (tsFileResourceToBeDeleted.isSeq()) {
+ deviceTimeIndexer = IndexerManager.getInstance().getSeqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
+ } else {
+ deviceTimeIndexer = IndexerManager.getInstance().getUnseqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
+ }
+ if (deviceTimeIndexer != null) {
+ deviceTimeIndexer.deleteIndexForDevices(tsFileResourceToBeDeleted.getDeviceToIndexMap(),
+ tsFileResourceToBeDeleted.getStartTimes(), tsFileResourceToBeDeleted.getEndTimes(),
+ tsFileResourceToBeDeleted.getTsFilePath());
+ }
+ }
logger.info("Delete tsfile {} successfully.", tsFileResourceToBeDeleted.getTsFile());
} finally {
tsFileResourceToBeDeleted.writeUnlock();
@@ -2532,4 +2617,34 @@ public class StorageGroupProcessor {
List<FlushListener> customFlushListeners) {
this.customFlushListeners = customFlushListeners;
}
+
+ public List<TsFileResource> getUnSealedListResourceForQuery(
+ Collection<TsFileResource> tsFileResources, PartialPath deviceId, String measurementId,
+ QueryContext context, Filter timeFilter, boolean isSeq) throws MetadataException {
+ MeasurementSchema schema = IoTDB.metaManager.getSeriesSchema(deviceId, measurementId);
+
+ List<TsFileResource> tsfileResourcesForQuery = new ArrayList<>();
+ long timeLowerBound = dataTTL != Long.MAX_VALUE ? System.currentTimeMillis() - dataTTL : Long
+ .MIN_VALUE;
+ context.setQueryTimeLowerBound(timeLowerBound);
+
+ for (TsFileResource tsFileResource : tsFileResources) {
+ if (!isTsFileResourceSatisfied(tsFileResource, deviceId.getFullPath(), timeFilter, isSeq)) {
+ continue;
+ }
+ closeQueryLock.readLock().lock();
+ try {
+ if (!tsFileResource.isClosed()) {
+ tsFileResource.getUnsealedFileProcessor()
+ .query(deviceId.getFullPath(), measurementId, schema.getType(), schema.getEncodingType(),
+ schema.getProps(), context, tsfileResourcesForQuery);
+ }
+ } catch (IOException e) {
+ throw new MetadataException(e);
+ } finally {
+ closeQueryLock.readLock().unlock();
+ }
+ }
+ return tsfileResourcesForQuery;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 4ed4d31..8711bd5 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -59,6 +59,8 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.rescon.PrimitiveArrayManager;
import org.apache.iotdb.db.rescon.SystemInfo;
import org.apache.iotdb.db.utils.MemUtils;
+import org.apache.iotdb.db.timeIndex.device.DeviceTimeIndexer;
+import org.apache.iotdb.db.timeIndex.IndexerManager;
import org.apache.iotdb.db.utils.QueryUtils;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.iotdb.db.writelog.WALFlushListener;
@@ -812,6 +814,16 @@ public class TsFileProcessor {
private void endFile() throws IOException, TsFileProcessorException {
long closeStartTime = System.currentTimeMillis();
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableDeviceIndexer()) {
+ // update device index
+ if (sequence) {
+ DeviceTimeIndexer seqIndexer = IndexerManager.getInstance().getSeqIndexer(storageGroupName);
+ seqIndexer.addIndexForDevices(tsFileResource.deviceToIndex, tsFileResource.startTimes, tsFileResource.endTimes, tsFileResource.getTsFilePath());
+ } else {
+ DeviceTimeIndexer unseqIndexer = IndexerManager.getInstance().getUnseqIndexer(storageGroupName);
+ unseqIndexer.addIndexForDevices(tsFileResource.deviceToIndex, tsFileResource.startTimes, tsFileResource.endTimes, tsFileResource.getTsFilePath());
+ }
+ }
tsFileResource.serialize();
writer.endFile();
tsFileResource.cleanCloseFlag();
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index 58f37b3..66d42d9 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -46,6 +46,7 @@ import org.apache.iotdb.db.engine.upgrade.UpgradeTask;
import org.apache.iotdb.db.exception.PartitionViolationException;
import org.apache.iotdb.db.rescon.CachedStringPool;
import org.apache.iotdb.db.service.UpgradeSevice;
+import org.apache.iotdb.db.timeIndex.device.DeviceTimeIndexer;
import org.apache.iotdb.db.utils.FilePathUtils;
import org.apache.iotdb.db.utils.UpgradeUtils;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
@@ -166,6 +167,10 @@ public class TsFileResource {
*/
private long minPlanIndex = Long.MAX_VALUE;
+ private DeviceTimeIndexer deviceTimeIndexer;
+ private String storageGroupName;
+
+
public TsFileResource() {
}
@@ -185,6 +190,7 @@ public class TsFileResource {
this.tsFileLock = other.tsFileLock;
this.fsFactory = other.fsFactory;
this.historicalVersions = other.historicalVersions;
+ this.storageGroupName = other.storageGroupName;
}
/**
@@ -210,6 +216,7 @@ public class TsFileResource {
initTimes(startTimes, Long.MAX_VALUE);
initTimes(endTimes, Long.MIN_VALUE);
this.processor = processor;
+ this.storageGroupName = processor.getStorageGroupName();
}
/**
@@ -227,6 +234,7 @@ public class TsFileResource {
this.readOnlyMemChunk = readOnlyMemChunk;
this.originTsFileResource = originTsFileResource;
generateTimeSeriesMetadata();
+ this.storageGroupName = originTsFileResource.getStorageGroupName();
}
private void generateTimeSeriesMetadata() throws IOException {
@@ -511,6 +519,9 @@ public class TsFileResource {
}
public void close() throws IOException {
+ //TODO update index
+
+
closed = true;
if (modFile != null) {
modFile.close();
@@ -868,7 +879,7 @@ public class TsFileResource {
if (file.exists()) {
Files.delete(file.toPath());
Files.delete(FSFactoryProducer.getFSFactory()
- .getFile(file.toPath() + TsFileResource.RESOURCE_SUFFIX).toPath());
+ .getFile(file.toPath() + TsFileResource.RESOURCE_SUFFIX).toPath());
}
}
@@ -895,6 +906,10 @@ public class TsFileResource {
public boolean isPlanIndexOverlap(TsFileResource another) {
return another.maxPlanIndex >= this.minPlanIndex &&
- another.minPlanIndex <= this.maxPlanIndex;
+ another.minPlanIndex <= this.maxPlanIndex;
+ }
+
+ public String getStorageGroupName() {
+ return storageGroupName;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index ea828c2..ccf31b0 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -127,6 +127,7 @@ import org.apache.iotdb.db.query.dataset.SingleDataSet;
import org.apache.iotdb.db.query.executor.IQueryRouter;
import org.apache.iotdb.db.query.executor.QueryRouter;
import org.apache.iotdb.db.service.IoTDB;
+import org.apache.iotdb.db.timeIndex.IndexerManager;
import org.apache.iotdb.db.utils.AuthUtils;
import org.apache.iotdb.db.utils.FileLoaderUtils;
import org.apache.iotdb.db.utils.UpgradeUtils;
@@ -139,6 +140,7 @@ import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.operator.In;
import org.apache.iotdb.tsfile.read.query.dataset.EmptyDataSet;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.iotdb.tsfile.utils.Binary;
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceIndex.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceIndex.java
deleted file mode 100644
index 8e37f8d..0000000
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceIndex.java
+++ /dev/null
@@ -1,4 +0,0 @@
-package org.apache.iotdb.db.timeIndex;
-
-public class DeviceIndex {
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceTimeIndexer.java
deleted file mode 100644
index 8ca9dc2..0000000
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/DeviceTimeIndexer.java
+++ /dev/null
@@ -1,4 +0,0 @@
-package org.apache.iotdb.db.timeIndex;
-
-public interface TimeIndexer {
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexConstants.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexConstants.java
index 296bb57..8c41375 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexConstants.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexConstants.java
@@ -1,4 +1,46 @@
+/*
+ * 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.timeIndex;
public class IndexConstants {
+ /**
+ * Indexer type
+ */
+ public enum IndexType {
+ // DeviceIndexer, query by device
+ DEVICE
+ }
+
+ /**
+ * the sub type of Indexer, we may has different implement of Indexer
+ */
+ public enum IndexSubType {
+
+ /**
+ * sub type of device Indexer
+ */
+ // load all file
+ LOADALL,
+ // use rocksdb to store the index
+ ROCKSDB
+ }
+
+ public static final String INDEXER_FILE = "indexer.snapshot";
+
}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java
index 4ce6e38..504dfd4 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/IndexerManager.java
@@ -1,4 +1,145 @@
+/*
+ * 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.timeIndex;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.timeIndex.device.DeviceTimeIndexer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
public class IndexerManager {
+ private String indexerFilePath;
+ private Map<PartialPath, DeviceTimeIndexer> seqIndexers;
+ private Map<PartialPath, DeviceTimeIndexer> unseqIndexers;
+ private ReentrantReadWriteLock lock;
+ private static final Logger logger = LoggerFactory.getLogger(IndexerManager.class);
+
+ private static class IndexerManagerHolder {
+
+ private IndexerManagerHolder() {
+ // allowed to do nothing
+ }
+
+ private static final IndexerManager INSTANCE = new IndexerManager();
+ }
+
+ public static IndexerManager getInstance() {
+ return IndexerManagerHolder.INSTANCE;
+ }
+
+ private IndexerManager() {
+ indexerFilePath = IoTDBDescriptor.getInstance().getConfig().getSchemaDir()
+ + File.pathSeparator + IndexConstants.INDEXER_FILE;
+ seqIndexers = new ConcurrentHashMap<>();
+ unseqIndexers = new ConcurrentHashMap<>();
+ lock = new ReentrantReadWriteLock();
+ }
+
+ /**
+ * init all indexer
+ * @return whether success
+ */
+ public boolean init() {
+ //TODO
+ // 1. get all storage group from file
+ // 2. init indexer for the storage group
+ return true;
+ }
+
+ public void addSeqIndexer(PartialPath storageGroup, DeviceTimeIndexer deviceTimeIndexer) {
+ lock.writeLock().lock();
+ seqIndexers.put(storageGroup, deviceTimeIndexer);
+ lock.writeLock().unlock();
+ }
+
+ public void addUnseqIndexer(PartialPath storageGroup, DeviceTimeIndexer deviceTimeIndexer) {
+ lock.writeLock().lock();
+ unseqIndexers.put(storageGroup, deviceTimeIndexer);
+ lock.writeLock().unlock();
+ }
+
+ public void deleteSeqIndexer(PartialPath storageGroup) {
+ lock.writeLock().lock();
+ seqIndexers.remove(storageGroup);
+ lock.writeLock().unlock();
+ }
+
+ public void deleteUnseqIndexer(PartialPath storageGroup) {
+ lock.writeLock().lock();
+ unseqIndexers.remove(storageGroup);
+ lock.writeLock().unlock();
+ }
+
+ public DeviceTimeIndexer getSeqIndexer(PartialPath storageGroup) {
+ lock.readLock().lock();
+ try {
+ return seqIndexers.get(storageGroup);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public DeviceTimeIndexer getSeqIndexer(String storageGroup) {
+ PartialPath sgName;
+ try {
+ sgName = new PartialPath(storageGroup);
+ } catch (IllegalPathException e) {
+ logger.warn("Fail to get DeviceTimeIndexer for storage group {}, err:{}", storageGroup, e.getMessage());
+ return null;
+ }
+ lock.readLock().lock();
+ try {
+ return seqIndexers.get(sgName);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public DeviceTimeIndexer getUnseqIndexer(PartialPath storageGroup) {
+ lock.readLock().lock();
+ try {
+ return unseqIndexers.get(storageGroup);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public DeviceTimeIndexer getUnseqIndexer(String storageGroup) {
+ PartialPath sgName;
+ try {
+ sgName = new PartialPath(storageGroup);
+ } catch (IllegalPathException e) {
+ logger.warn("Fail to get DeviceTimeIndexer for storage group {}, err:{}", storageGroup, e.getMessage());
+ return null;
+ }
+ lock.readLock().lock();
+ try {
+ return unseqIndexers.get(sgName);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/LoadAllTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/LoadAllTimeIndexer.java
deleted file mode 100644
index c553213..0000000
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/LoadAllTimeIndexer.java
+++ /dev/null
@@ -1,4 +0,0 @@
-package org.apache.iotdb.db.timeIndex;
-
-public class LoadAllTimeIndexer {
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/RocksDBDeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/RocksDBDeviceTimeIndexer.java
deleted file mode 100644
index 219b7f0..0000000
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/RocksDBDeviceTimeIndexer.java
+++ /dev/null
@@ -1,4 +0,0 @@
-package org.apache.iotdb.db.timeIndex;
-
-public class RocksDBDeviceTimeIndexer {
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexer.java
new file mode 100644
index 0000000..a2d9d17
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexer.java
@@ -0,0 +1,104 @@
+package org.apache.iotdb.db.timeIndex;
+
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.timeIndex.device.UpdateIndexsParam;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+
+import java.util.List;
+import java.util.Map;
+
+public interface TimeIndexer {
+ /**
+ * init the Indexer when IoTDB start
+ * @return whether success
+ */
+ public boolean init();
+
+ /**
+ * may do some prepared work before operation
+ * @return whether success
+ */
+ public boolean begin();
+
+ /**
+ * may do some resource release after operation
+ * @return whether success
+ */
+ public boolean end();
+
+ /**
+ * add one index record for the path
+ * @param path
+ * @param startTime
+ * @param endTime
+ * @param tsFilePath
+ * @return
+ */
+ public boolean addIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath);
+
+ /**
+ * aadd one index record for the path
+ * @param path
+ * @param startTime
+ * @param endTime
+ * @param tsFilePath
+ * @return
+ */
+ public boolean addIndexForPath(String path, long startTime, long endTime, String tsFilePath);
+
+ /**
+ * add all indexs for a flushed tsFile
+ * @param paths
+ * @param startTimes
+ * @param endTimes
+ * @param tsFilePath
+ * @return
+ */
+ public boolean addIndexForPaths(Map<PartialPath, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath);
+
+ /**
+ * delete one index for the path
+ * @param path
+ * @param startTime
+ * @param endTime
+ * @param tsFilePath
+ * @return whether success
+ */
+ public boolean deleteIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath);
+
+ /**
+ * delete one index for the path
+ * @param path
+ * @param startTime
+ * @param endTime
+ * @param tsFilePath
+ * @return
+ */
+ public boolean deleteIndexForPath(String path, long startTime, long endTime, String tsFilePath);
+
+ /**
+ * delete all index for one deleted tsfile
+ * @param paths
+ * @param startTimes
+ * @param endTimes
+ * @param tsFilePath
+ * @return
+ */
+ public boolean deleteIndexForDevices(Map<PartialPath, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath);
+
+ /**
+ * after merge, we should keep the index is updated in consistency
+ * @param updateIndexsParam
+ * @return whether success
+ */
+ public boolean updateIndexForDevices(UpdateIndexsParam updateIndexsParam);
+
+ /**
+ * found the related tsFile(only cover sealed tsfile) for one deviceId
+ * @param path
+ * @param timeFilter
+ * @return whether success
+ */
+ public List<TsFileResource> filterByOneDevice(PartialPath path, Filter timeFilter);
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java
deleted file mode 100644
index 80a7845..0000000
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java
+++ /dev/null
@@ -1,4 +0,0 @@
-package org.apache.iotdb.db.timeIndex;
-
-public class UpdateIndexParam {
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceIndex.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceIndex.java
new file mode 100644
index 0000000..bafa0c1
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceIndex.java
@@ -0,0 +1,63 @@
+/*
+ * 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.timeIndex.device;
+
+import org.apache.iotdb.db.metadata.PartialPath;
+
+/**
+ * Device Index, like [(deviceId, startTime, endTime, TsFilePath)] to accelerate query
+ */
+public class DeviceIndex {
+ private PartialPath[] deviceIds;
+ private long[] startTimes;
+ private long[] endTimes;
+ private String tsFilePath;
+
+ public PartialPath[] getDeviceIds() {
+ return deviceIds;
+ }
+
+ public void setDeviceIds(PartialPath[] deviceIds) {
+ this.deviceIds = deviceIds;
+ }
+
+ public long[] getStartTimes() {
+ return startTimes;
+ }
+
+ public void setStartTimes(long[] startTimes) {
+ this.startTimes = startTimes;
+ }
+
+ public long[] getEndTimes() {
+ return endTimes;
+ }
+
+ public void setEndTimes(long[] endTimes) {
+ this.endTimes = endTimes;
+ }
+
+ public String getTsFilePath() {
+ return tsFilePath;
+ }
+
+ public void setTsFilePath(String tsFilePath) {
+ this.tsFilePath = tsFilePath;
+ }
+}
\ No newline at end of file
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceTimeIndexer.java
new file mode 100644
index 0000000..d846d0d
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/DeviceTimeIndexer.java
@@ -0,0 +1,124 @@
+/*
+ * 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.timeIndex.device;
+
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * manage the index [deviceId, (startTime, endTime, TsFilePath)] for all devices
+ */
+public interface DeviceTimeIndexer {
+ /**
+ * init the Indexer when IoTDB start
+ * @return whether success
+ */
+ public boolean init();
+
+ /**
+ * may do some prepared work before operation
+ * @return whether success
+ */
+ public boolean begin();
+
+ /**
+ * may do some resource release after operation
+ * @return whether success
+ */
+ public boolean end();
+
+ /**
+ * add one index for the deviceId
+ * @param deviceId
+ * @param startTime
+ * @param endTime
+ * @param tsFilePath
+ * @return whether success
+ */
+ public boolean addIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath);
+
+ /**
+ * add one index for the deviceId
+ * @param deviceId
+ * @param startTime
+ * @param endTime
+ * @param tsFilePath
+ * @return
+ */
+ public boolean addIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath);
+
+ /**
+ * add all indexs for a flushed tsFile
+ * @param deviceIds
+ * @param startTimes
+ * @param endTimes
+ * @param tsFilePath
+ * @return
+ */
+ public boolean addIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath);
+
+ /**
+ * delete one index for the deviceId
+ * @param deviceId
+ * @param startTime
+ * @param endTime
+ * @param tsFilePath
+ * @return whether success
+ */
+ public boolean deleteIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath);
+
+ /**
+ * delete one index for the deviceId
+ * @param deviceId
+ * @param startTime
+ * @param endTime
+ * @param tsFilePath
+ * @return
+ */
+ public boolean deleteIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath);
+
+ /**
+ * delete all index for one deleted tsfile
+ * @param deviceIds
+ * @param startTimes
+ * @param endTimes
+ * @param tsFilePath
+ * @return
+ */
+ public boolean deleteIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath);
+
+ /**
+ * after merge, we should keep the index is updated in consistency
+ * @param updateIndexsParam
+ * @return whether success
+ */
+ public boolean updateIndexForDevices(UpdateIndexsParam updateIndexsParam);
+
+ /**
+ * found the related tsFile(only cover sealed tsfile) for one deviceId
+ * @param deviceId
+ * @param timeFilter
+ * @return whether success
+ */
+ public List<TsFileResource> filterByOneDevice(PartialPath deviceId, Filter timeFilter);
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllTimeIndexer.java
new file mode 100644
index 0000000..c5dab56
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllTimeIndexer.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.timeIndex.device;
+
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+
+import java.util.List;
+import java.util.Map;
+
+public class LoadAllTimeIndexer implements DeviceTimeIndexer {
+
+ public LoadAllTimeIndexer() {
+
+ }
+
+ @Override
+ public boolean init() {
+ return false;
+ }
+
+ @Override
+ public boolean begin() {
+ return false;
+ }
+
+ @Override
+ public boolean end() {
+ return false;
+ }
+
+ @Override
+ public boolean addIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean addIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean addIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean deleteIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean deleteIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean deleteIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean updateIndexForDevices(UpdateIndexsParam updateIndexsParam) {
+ return false;
+ }
+
+ @Override
+ public List<TsFileResource> filterByOneDevice(PartialPath deviceId, Filter timeFilter) {
+ return null;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
new file mode 100644
index 0000000..62fa0cf
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
@@ -0,0 +1,83 @@
+/*
+ * 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.timeIndex.device;
+
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+
+import java.util.List;
+import java.util.Map;
+
+public class RocksDBDeviceTimeIndexer implements DeviceTimeIndexer {
+ @Override
+ public boolean init() {
+ return false;
+ }
+
+ @Override
+ public boolean begin() {
+ return false;
+ }
+
+ @Override
+ public boolean end() {
+ return false;
+ }
+
+ @Override
+ public boolean addIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean addIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean addIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean deleteIndexForDevice(PartialPath deviceId, long startTime, long endTime, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean deleteIndexForDevice(String deviceId, long startTime, long endTime, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean deleteIndexForDevices(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath) {
+ return false;
+ }
+
+ @Override
+ public boolean updateIndexForDevices(UpdateIndexsParam updateIndexsParam) {
+ return false;
+ }
+
+ @Override
+ public List<TsFileResource> filterByOneDevice(PartialPath deviceId, Filter timeFilter) {
+ return null;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/UpdateIndexsParam.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/UpdateIndexsParam.java
new file mode 100644
index 0000000..e99fc3f
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/UpdateIndexsParam.java
@@ -0,0 +1,43 @@
+/*
+ * 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.timeIndex.device;
+
+/**
+ * to atomic update indexs for device
+ */
+public class UpdateIndexsParam {
+ private DeviceIndex[] oldIndexs;
+ private DeviceIndex newIndex;
+
+ public DeviceIndex[] getOldIndexs() {
+ return oldIndexs;
+ }
+
+ public void setOldIndexs(DeviceIndex[] oldIndexs) {
+ this.oldIndexs = oldIndexs;
+ }
+
+ public DeviceIndex getNewIndex() {
+ return newIndex;
+ }
+
+ public void setNewIndex(DeviceIndex newIndex) {
+ this.newIndex = newIndex;
+ }
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
index 04ab1ba..87906af 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
@@ -19,14 +19,11 @@
package org.apache.iotdb.tsfile.utils;
-import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.BINARY;
-import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.BOOLEAN;
-import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.DOUBLE;
-import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.FLOAT;
-import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.INTEGER;
-import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.LONG;
-import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.NULL;
-import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.STRING;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.reader.TsFileInput;
import java.io.DataOutputStream;
import java.io.IOException;
@@ -34,17 +31,10 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
import java.util.Map.Entry;
-import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
-import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.iotdb.tsfile.read.reader.TsFileInput;
+
+import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.*;
/**
* ConverterUtils is a utility class. It provide conversion between normal datatype and byte array.
[iotdb] 05/05: refactor the index
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chaow pushed a commit to branch fileIndex
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1e67eaca438545b804cd8700a683a2b68a39c92c
Author: chaow <xu...@gmail.com>
AuthorDate: Mon Oct 12 11:47:11 2020 +0800
refactor the index
---
.../resources/conf/iotdb-engine.properties | 1 +
.../iotdb/db/engine/merge/task/MergeFileTask.java | 5 +-
.../engine/storagegroup/StorageGroupProcessor.java | 29 +++++----
.../db/engine/storagegroup/TsFileProcessor.java | 4 +-
.../db/engine/storagegroup/TsFileResource.java | 3 -
.../iotdb/db/timeIndex/FileIndexEntries.java | 67 +++++++++++++++++++++
.../iotdb/db/timeIndex/FileIndexerManager.java | 3 +
.../apache/iotdb/db/timeIndex/FileTimeIndexer.java | 68 +++-------------------
.../{TimeIndex.java => TimeIndexEntry.java} | 47 +++++++--------
.../iotdb/db/timeIndex/UpdateIndexsParam.java | 43 --------------
.../timeIndex/device/LoadAllDeviceTimeIndexer.java | 35 ++---------
.../timeIndex/device/RocksDBDeviceTimeIndexer.java | 35 ++---------
12 files changed, 128 insertions(+), 212 deletions(-)
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties b/server/src/assembly/resources/conf/iotdb-engine.properties
index f8b2ca4..54184e1 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -254,6 +254,7 @@ max_waiting_time_when_insert_blocked=0
# estimated metadata size (in byte) of one timeseries in Mtree
estimated_series_size=300
+# time indexer
enable_file_time_indexer=false
# 0, device indexer base on file; 1, device indexer base on rocksdb
file_time_indexer_type=0
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
index a806271..d0e2efd 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
@@ -41,6 +41,7 @@ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.query.control.FileReaderManager;
+import org.apache.iotdb.db.timeIndex.FileIndexEntries;
import org.apache.iotdb.db.timeIndex.FileIndexerManager;
import org.apache.iotdb.db.timeIndex.FileTimeIndexer;
import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
@@ -204,9 +205,9 @@ class MergeFileTask {
if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
// add new device index
- // may Indexer need delete old index background
+ // may Indexer need delete old index background, or just overwrite old index
FileTimeIndexer fileTimeIndexer = FileIndexerManager.getInstance().getSeqIndexer(seqFile.getStorageGroupName());
- fileTimeIndexer.addIndexForPaths(seqFile.getDeviceToIndexMap(), seqFile.getStartTimes(), seqFile.getEndTimes(), seqFile.getTsFilePath());
+ fileTimeIndexer.addIndexForPaths(FileIndexEntries.convertFromTsFileResource(seqFile));
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index 2e03646..d6f6ef8 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -82,8 +82,10 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryFileManager;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.service.UpgradeSevice;
+import org.apache.iotdb.db.timeIndex.FileIndexEntries;
import org.apache.iotdb.db.timeIndex.FileIndexerManager;
import org.apache.iotdb.db.timeIndex.FileTimeIndexer;
+import org.apache.iotdb.db.timeIndex.TimeIndexEntry;
import org.apache.iotdb.db.utils.CopyOnReadLinkedList;
import org.apache.iotdb.db.writelog.recover.TsFileRecoverPerformer;
import org.apache.iotdb.rpc.RpcUtils;
@@ -1304,15 +1306,11 @@ public class StorageGroupProcessor {
tsFileManagement.remove(resource, isSeq);
if (isSeq) {
if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
- FileTimeIndexer fileTimeIndexer = FileIndexerManager.getInstance().getSeqIndexer(resource.getStorageGroupName());
- fileTimeIndexer.deleteIndexForPaths(resource.getDeviceToIndexMap(), resource.getStartTimes(),
- resource.getEndTimes(), resource.getTsFilePath());
+ seqFileTimeIndexer.deleteIndexForPaths(FileIndexEntries.convertFromTsFileResource(resource));
}
} else {
if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
- FileTimeIndexer fileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(resource.getStorageGroupName());
- fileTimeIndexer.deleteIndexForPaths(resource.getDeviceToIndexMap(), resource.getStartTimes(),
- resource.getEndTimes(), resource.getTsFilePath());
+ unseqFileTimeIndexer.deleteIndexForPaths(FileIndexEntries.convertFromTsFileResource(resource));
}
}
} catch (IllegalPathException e) {
@@ -1404,8 +1402,13 @@ public class StorageGroupProcessor {
List<TsFileResource> unseqResources;
try {
if (IoTDBDescriptor.getInstance().getConfig().isEnableFileTimeIndexer()) {
- seqResources = seqFileTimeIndexer.filterByPath(deviceId, timeFilter);
- unseqResources = unseqFileTimeIndexer.filterByPath(deviceId, timeFilter);
+ List<FileIndexEntries> seqFiles = seqFileTimeIndexer.filterByPath(deviceId, timeFilter);
+ List<FileIndexEntries> unseqFiles = unseqFileTimeIndexer.filterByPath(deviceId, timeFilter);
+
+ // TODO load resource or just use resource according to the implements
+ seqResources = new ArrayList<>(seqFiles.size());
+ unseqResources = new ArrayList<>(unseqFiles.size());
+
List<TsFileResource> unsealedSeqFiles = getUnSealedListResourceForQuery(
tsFileManagement.getTsFileList(true),
deviceId, measurementId, context, timeFilter, true);
@@ -1902,8 +1905,7 @@ public class StorageGroupProcessor {
} else {
fileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
}
- fileTimeIndexer.addIndexForPaths(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
- newTsFileResource.getEndTimes(), newTsFileResource.getTsFilePath());
+ fileTimeIndexer.addIndexForPaths(FileIndexEntries.convertFromTsFileResource(newTsFileResource));
}
} catch (DiskSpaceInsufficientException | IllegalPathException e) {
logger.error(
@@ -1973,8 +1975,7 @@ public class StorageGroupProcessor {
} else {
fileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(newTsFileResource.getStorageGroupName());
}
- fileTimeIndexer.addIndexForPaths(newTsFileResource.getDeviceToIndexMap(), newTsFileResource.getStartTimes(),
- newTsFileResource.getEndTimes(), newTsFileResource.getTsFilePath());
+ fileTimeIndexer.addIndexForPaths(FileIndexEntries.convertFromTsFileResource(newTsFileResource));
} catch (IllegalPathException e) {
logger.error("Fail to get DeviceTimeIndexer for storage group {}, err:{}", newTsFileResource.getStorageGroupName(), e.getMessage());
IoTDBDescriptor.getInstance().getConfig().setReadOnly(true);
@@ -2391,9 +2392,7 @@ public class StorageGroupProcessor {
fileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(tsFileResourceToBeDeleted.getStorageGroupName());
}
if (fileTimeIndexer != null) {
- fileTimeIndexer.deleteIndexForPaths(tsFileResourceToBeDeleted.getDeviceToIndexMap(),
- tsFileResourceToBeDeleted.getStartTimes(), tsFileResourceToBeDeleted.getEndTimes(),
- tsFileResourceToBeDeleted.getTsFilePath());
+ fileTimeIndexer.deleteIndexForPaths(FileIndexEntries.convertFromTsFileResource(tsFileResourceToBeDeleted));
}
} catch (IllegalPathException e) {
return false;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index cf18c83..bb146f3 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -60,6 +60,7 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.rescon.PrimitiveArrayManager;
import org.apache.iotdb.db.rescon.SystemInfo;
import org.apache.iotdb.db.utils.MemUtils;
+import org.apache.iotdb.db.timeIndex.FileIndexEntries;
import org.apache.iotdb.db.timeIndex.FileIndexerManager;
import org.apache.iotdb.db.timeIndex.FileTimeIndexer;
import org.apache.iotdb.db.utils.QueryUtils;
@@ -824,8 +825,7 @@ public class TsFileProcessor {
} else {
fileTimeIndexer = FileIndexerManager.getInstance().getUnseqIndexer(storageGroupName);
}
- fileTimeIndexer.addIndexForPaths(tsFileResource.deviceToIndex, tsFileResource.startTimes,
- tsFileResource.endTimes, tsFileResource.getTsFilePath());
+ fileTimeIndexer.addIndexForPaths(FileIndexEntries.convertFromTsFileResource(tsFileResource));
} catch (IllegalPathException e) {
logger.error("Failed to endFile {} for storage group {}, err:{}", tsFileResource.getTsFile(),
tsFileResource.getStorageGroupName(), e.getMessage());
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index ca2b819..c55c73e 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -517,9 +517,6 @@ public class TsFileResource {
}
public void close() throws IOException {
- //TODO update index
-
-
closed = true;
if (modFile != null) {
modFile.close();
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/FileIndexEntries.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/FileIndexEntries.java
new file mode 100644
index 0000000..8f566ba
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/FileIndexEntries.java
@@ -0,0 +1,67 @@
+/*
+ * 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.timeIndex;
+
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.PartialPath;
+
+import java.util.Map;
+
+/**
+ * Time Index, like deviceIndex: [(deviceId, startTime, endTime, TsFilePath)] to accelerate query
+ */
+public class FileIndexEntries {
+ private TimeIndexEntry[] indexEntries;
+ private String tsFilePath;
+
+ public TimeIndexEntry[] getIndexEntries() {
+ return indexEntries;
+ }
+
+ public void setIndexEntries(TimeIndexEntry[] indexEntries) {
+ this.indexEntries = indexEntries;
+ }
+
+ public String getTsFilePath() {
+ return tsFilePath;
+ }
+
+ public void setTsFilePath(String tsFilePath) {
+ this.tsFilePath = tsFilePath;
+ }
+
+ public static FileIndexEntries convertFromTsFileResource(TsFileResource resource) throws IllegalPathException {
+ FileIndexEntries fileIndexEntries = new FileIndexEntries();
+ TimeIndexEntry[] timeIndexEntries = new TimeIndexEntry[resource.getDeviceToIndexMap().size()];
+ int i = 0;
+ for (Map.Entry<String, Integer> entry : resource.getDeviceToIndexMap().entrySet()) {
+ TimeIndexEntry timeIndexEntry = new TimeIndexEntry();
+ timeIndexEntry.setAllElem(
+ new PartialPath(entry.getKey()),
+ resource.getStartTime(entry.getValue()),
+ resource.getEndTime(entry.getValue()));
+ timeIndexEntries[i++] = timeIndexEntry;
+ }
+ fileIndexEntries.setIndexEntries(timeIndexEntries);
+ fileIndexEntries.setTsFilePath(resource.getTsFilePath());
+
+ return fileIndexEntries;
+ }
+}
\ No newline at end of file
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/FileIndexerManager.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/FileIndexerManager.java
index 2c9f602..caeed74 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/FileIndexerManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/FileIndexerManager.java
@@ -29,6 +29,9 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+/**
+ * Manage all indexers
+ */
public class FileIndexerManager {
private String indexerFilePath;
private Map<PartialPath, FileTimeIndexer> seqIndexers;
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/FileTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/FileTimeIndexer.java
index 64c6dab..5172e92 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/FileTimeIndexer.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/FileTimeIndexer.java
@@ -18,7 +18,6 @@
*/
package org.apache.iotdb.db.timeIndex;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
@@ -45,77 +44,26 @@ public interface FileTimeIndexer {
public boolean end();
/**
- * add one index record for the path
- * @param path
- * @param startTime
- * @param endTime
- * @param tsFilePath
- * @return
- */
- public boolean addIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath);
-
- /**
- * aadd one index record for the path
- * @param path
- * @param startTime
- * @param endTime
- * @param tsFilePath
- * @return
- */
- public boolean addIndexForPath(String path, long startTime, long endTime, String tsFilePath);
-
- /**
* add all indexs for a flushed tsFile
- * @param paths
- * @param startTimes
- * @param endTimes
- * @param tsFilePath
+ *
+ * @param fileIndexEntries
* @return
*/
- public boolean addIndexForPaths(Map<String, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath);
+ public boolean addIndexForPaths(FileIndexEntries fileIndexEntries);
/**
* delete one index for the path
- * @param path
- * @param startTime
- * @param endTime
- * @param tsFilePath
- * @return whether success
- */
- public boolean deleteIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath);
-
- /**
- * delete one index for the path
- * @param path
- * @param startTime
- * @param endTime
- * @param tsFilePath
- * @return
- */
- public boolean deleteIndexForPath(String path, long startTime, long endTime, String tsFilePath);
-
- /**
- * delete all index for one deleted tsfile
- * @param paths
- * @param startTimes
- * @param endTimes
- * @param tsFilePath
+ *
+ * @param fileIndexEntries
* @return
*/
- public boolean deleteIndexForPaths(Map<String, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath);
-
- /**
- * after merge, we should keep the index is updated in consistency
- * @param updateIndexsParam
- * @return whether success
- */
- public boolean updateIndexForPaths(UpdateIndexsParam updateIndexsParam);
+ public boolean deleteIndexForPaths(FileIndexEntries fileIndexEntries);
/**
* found the related tsFile(only cover sealed tsfile) for one deviceId
- * @param path
+ * @param path does not support regex match
* @param timeFilter
* @return whether success
*/
- public List<TsFileResource> filterByPath(PartialPath path, Filter timeFilter);
+ public List<FileIndexEntries> filterByPath(PartialPath path, Filter timeFilter);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndex.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexEntry.java
similarity index 54%
rename from server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndex.java
rename to server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexEntry.java
index 5cf3705..0b9d385 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndex.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/TimeIndexEntry.java
@@ -21,43 +21,40 @@ package org.apache.iotdb.db.timeIndex;
import org.apache.iotdb.db.metadata.PartialPath;
/**
- * Time Index, like deviceIndex: [(deviceId, startTime, endTime, TsFilePath)] to accelerate query
+ * time index entry, contains path, startTime, endTime
*/
-public class TimeIndex {
- private PartialPath[] paths;
- private long[] startTimes;
- private long[] endTimes;
- private String tsFilePath;
-
- public PartialPath[] getPaths() {
- return paths;
- }
+public class TimeIndexEntry {
+ private PartialPath path;
+ private long startTime;
+ private long endTime;
- public void setPaths(PartialPath[] paths) {
- this.paths = paths;
+ public PartialPath getPath() {
+ return path;
}
- public long[] getStartTimes() {
- return startTimes;
+ public void setPath(PartialPath path) {
+ this.path = path;
}
- public void setStartTimes(long[] startTimes) {
- this.startTimes = startTimes;
+ public long getStartTime() {
+ return startTime;
}
- public long[] getEndTimes() {
- return endTimes;
+ public void setStartTime(long startTime) {
+ this.startTime = startTime;
}
- public void setEndTimes(long[] endTimes) {
- this.endTimes = endTimes;
+ public long getEndTime() {
+ return endTime;
}
- public String getTsFilePath() {
- return tsFilePath;
+ public void setEndTime(long endTime) {
+ this.endTime = endTime;
}
- public void setTsFilePath(String tsFilePath) {
- this.tsFilePath = tsFilePath;
+ public void setAllElem(PartialPath path, long startTime, long endTime) {
+ this.path = path;
+ this.startTime = startTime;
+ this.endTime = endTime;
}
-}
\ No newline at end of file
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java
deleted file mode 100644
index e7b14b7..0000000
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/UpdateIndexsParam.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.db.timeIndex;
-
-/**
- * to atomic update indexs for path
- */
-public class UpdateIndexsParam {
- private TimeIndex[] oldIndexs;
- private TimeIndex newIndex;
-
- public TimeIndex[] getOldIndexs() {
- return oldIndexs;
- }
-
- public void setOldIndexs(TimeIndex[] oldIndexs) {
- this.oldIndexs = oldIndexs;
- }
-
- public TimeIndex getNewIndex() {
- return newIndex;
- }
-
- public void setNewIndex(TimeIndex newIndex) {
- this.newIndex = newIndex;
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllDeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllDeviceTimeIndexer.java
index 3d02b9c..a15bdef 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllDeviceTimeIndexer.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/LoadAllDeviceTimeIndexer.java
@@ -18,14 +18,12 @@
*/
package org.apache.iotdb.db.timeIndex.device;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.timeIndex.FileIndexEntries;
import org.apache.iotdb.db.timeIndex.FileTimeIndexer;
-import org.apache.iotdb.db.timeIndex.UpdateIndexsParam;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.util.List;
-import java.util.Map;
public class LoadAllDeviceTimeIndexer implements FileTimeIndexer {
@@ -49,42 +47,17 @@ public class LoadAllDeviceTimeIndexer implements FileTimeIndexer {
}
@Override
- public boolean addIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath) {
+ public boolean addIndexForPaths(FileIndexEntries fileIndexEntries) {
return false;
}
@Override
- public boolean addIndexForPath(String path, long startTime, long endTime, String tsFilePath) {
+ public boolean deleteIndexForPaths(FileIndexEntries fileIndexEntries) {
return false;
}
@Override
- public boolean addIndexForPaths(Map<String, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath) {
- return false;
- }
-
- @Override
- public boolean deleteIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath) {
- return false;
- }
-
- @Override
- public boolean deleteIndexForPath(String path, long startTime, long endTime, String tsFilePath) {
- return false;
- }
-
- @Override
- public boolean deleteIndexForPaths(Map<String, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath) {
- return false;
- }
-
- @Override
- public boolean updateIndexForPaths(UpdateIndexsParam updateIndexsParam) {
- return false;
- }
-
- @Override
- public List<TsFileResource> filterByPath(PartialPath path, Filter timeFilter) {
+ public List<FileIndexEntries> filterByPath(PartialPath path, Filter timeFilter) {
return null;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
index 686531f..029bb62 100644
--- a/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
+++ b/server/src/main/java/org/apache/iotdb/db/timeIndex/device/RocksDBDeviceTimeIndexer.java
@@ -18,14 +18,12 @@
*/
package org.apache.iotdb.db.timeIndex.device;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.timeIndex.FileIndexEntries;
import org.apache.iotdb.db.timeIndex.FileTimeIndexer;
-import org.apache.iotdb.db.timeIndex.UpdateIndexsParam;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.util.List;
-import java.util.Map;
public class RocksDBDeviceTimeIndexer implements FileTimeIndexer {
@Override
@@ -44,42 +42,17 @@ public class RocksDBDeviceTimeIndexer implements FileTimeIndexer {
}
@Override
- public boolean addIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath) {
+ public boolean addIndexForPaths(FileIndexEntries fileIndexEntries) {
return false;
}
@Override
- public boolean addIndexForPath(String path, long startTime, long endTime, String tsFilePath) {
+ public boolean deleteIndexForPaths(FileIndexEntries fileIndexEntries) {
return false;
}
@Override
- public boolean addIndexForPaths(Map<String, Integer> paths, long[] startTimes, long[] endTimes, String tsFilePath) {
- return false;
- }
-
- @Override
- public boolean deleteIndexForPath(PartialPath path, long startTime, long endTime, String tsFilePath) {
- return false;
- }
-
- @Override
- public boolean deleteIndexForPath(String path, long startTime, long endTime, String tsFilePath) {
- return false;
- }
-
- @Override
- public boolean deleteIndexForPaths(Map<String, Integer> deviceIds, long[] startTimes, long[] endTimes, String tsFilePath) {
- return false;
- }
-
- @Override
- public boolean updateIndexForPaths(UpdateIndexsParam updateIndexsParam) {
- return false;
- }
-
- @Override
- public List<TsFileResource> filterByPath(PartialPath path, Filter timeFilter) {
+ public List<FileIndexEntries> filterByPath(PartialPath path, Filter timeFilter) {
return null;
}
}