This is an automated email from the ASF dual-hosted git repository. geniuspig pushed a commit to branch optimize_path_v2 in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 8a30a4fcf30cd497b799a4bf145948152de7615c Author: zhutianci <[email protected]> AuthorDate: Sat Aug 15 16:24:36 2020 +0800 excutor and physical --- .../org/apache/iotdb/db/engine/StorageEngine.java | 79 +++++------ .../iotdb/db/engine/modification/Deletion.java | 5 +- .../iotdb/db/engine/modification/Modification.java | 15 +- .../db/engine/querycontext/QueryDataSource.java | 7 +- .../engine/storagegroup/StorageGroupProcessor.java | 55 +++++--- .../db/engine/storagegroup/TsFileProcessor.java | 8 +- .../org/apache/iotdb/db/metadata/MManager.java | 151 +++++++++++---------- .../java/org/apache/iotdb/db/metadata/MTree.java | 44 +++--- .../apache/iotdb/db/qp/executor/IPlanExecutor.java | 3 +- .../apache/iotdb/db/qp/executor/PlanExecutor.java | 128 ++++++++--------- .../db/qp/logical/crud/BasicFunctionOperator.java | 13 +- .../db/qp/logical/crud/BasicOperatorType.java | 1 + .../iotdb/db/qp/logical/crud/InOperator.java | 12 +- .../db/qp/physical/crud/DeletePartitionPlan.java | 10 +- .../db/query/control/QueryResourceManager.java | 4 +- .../groupby/GroupByWithValueFilterDataSet.java | 7 +- .../groupby/GroupByWithoutValueFilterDataSet.java | 7 +- .../dataset/groupby/LocalGroupByExecutor.java | 3 +- .../iotdb/db/query/executor/IQueryRouter.java | 5 +- .../iotdb/db/query/executor/QueryRouter.java | 47 ++++--- .../iotdb/db/sync/receiver/load/FileLoader.java | 3 +- 21 files changed, 326 insertions(+), 281 deletions(-) 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 57f97ff..9c72e4e 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 @@ -32,10 +32,12 @@ import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartiti import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor; import org.apache.iotdb.db.engine.storagegroup.TsFileResource; import org.apache.iotdb.db.exception.*; +import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException; +import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode; import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan; import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan; @@ -75,7 +77,7 @@ public class StorageEngine implements IService { /** * storage group name -> storage group processor */ - private final ConcurrentHashMap<String, StorageGroupProcessor> processorMap = new ConcurrentHashMap<>(); + private final ConcurrentHashMap<PartialPath, StorageGroupProcessor> processorMap = new ConcurrentHashMap<>(); private static final ExecutorService recoveryThreadPool = IoTDBThreadPoolFactory .newFixedThreadPool(Runtime.getRuntime().availableProcessors(), "Recovery-Thread-Pool"); @@ -163,7 +165,7 @@ public class StorageEngine implements IService { StorageGroupProcessor processor = new StorageGroupProcessor(systemDir, storageGroup.getFullPath(), fileFlushPolicy); processor.setDataTTL(storageGroup.getDataTTL()); - processorMap.put(storageGroup.getFullPath(), processor); + processorMap.put(storageGroup.getPartialPath(), processor); logger.info("Storage Group Processor {} is recovered successfully", storageGroup.getFullPath()); } catch (Exception e) { @@ -270,31 +272,30 @@ public class StorageEngine implements IService { return ServiceType.STORAGE_ENGINE_SERVICE; } - public StorageGroupProcessor getProcessor(String path) throws StorageEngineException { - String storageGroupName; + public StorageGroupProcessor getProcessor(PartialPath path) throws StorageEngineException { + PartialPath storageGroup; try { - storageGroupName = IoTDB.metaManager.getStorageGroupName(path); + storageGroup = IoTDB.metaManager.getStorageGroupName(path); StorageGroupProcessor processor; - processor = processorMap.get(storageGroupName); + processor = processorMap.get(storageGroup); if (processor == null) { // if finish recover if (isAllSgReady.get()) { - storageGroupName = storageGroupName.intern(); - synchronized (storageGroupName) { - processor = processorMap.get(storageGroupName); + synchronized (storageGroup) { + processor = processorMap.get(storageGroup); if (processor == null) { logger.info("construct a processor instance, the storage group is {}, Thread is {}", - storageGroupName, Thread.currentThread().getId()); - processor = new StorageGroupProcessor(systemDir, storageGroupName, fileFlushPolicy); - StorageGroupMNode storageGroup = IoTDB.metaManager - .getStorageGroupNode(storageGroupName); - processor.setDataTTL(storageGroup.getDataTTL()); - processorMap.put(storageGroupName, processor); + storageGroup, Thread.currentThread().getId()); + processor = new StorageGroupProcessor(systemDir, storageGroup.toString(), fileFlushPolicy); + StorageGroupMNode storageGroupMNode = IoTDB.metaManager + .getStorageGroupNode(storageGroup); + processor.setDataTTL(storageGroupMNode.getDataTTL()); + processorMap.put(storageGroup, processor); } } } else { // not finished recover, refuse the request - throw new StorageEngineException("the sg " + storageGroupName + " may not ready now, please wait and retry later", + throw new StorageEngineException("the sg " + storageGroup + " may not ready now, please wait and retry later", TSStatusCode.STORAGE_GROUP_NOT_READY.getStatusCode()); } } @@ -430,7 +431,7 @@ public class StorageEngine implements IService { /** * delete data of timeseries "{deviceId}.{measurementId}" with time <= timestamp. */ - public void delete(String deviceId, String measurementId, long startTime, long endTime) + public void delete(PartialPath deviceId, String measurementId, long startTime, long endTime) throws StorageEngineException { StorageGroupProcessor storageGroupProcessor = getProcessor(deviceId); try { @@ -445,8 +446,8 @@ public class StorageEngine implements IService { */ public QueryDataSource query(SingleSeriesExpression seriesExpression, QueryContext context, QueryFileManager filePathsManager) - throws StorageEngineException, QueryProcessException { - String deviceId = seriesExpression.getSeriesPath().getDevice(); + throws StorageEngineException, QueryProcessException, IllegalPathException { + PartialPath deviceId = new PartialPath(seriesExpression.getSeriesPath().getDevice()); String measurementId = seriesExpression.getSeriesPath().getMeasurement(); StorageGroupProcessor storageGroupProcessor = getProcessor(deviceId); return storageGroupProcessor @@ -499,13 +500,13 @@ public class StorageEngine implements IService { * delete all data files (both memory data and file on disk) in a storage group. It is used when * there is no timeseries (which are all deleted) in this storage group) */ - public void deleteAllDataFilesInOneStorageGroup(String storageGroupName) { + public void deleteAllDataFilesInOneStorageGroup(PartialPath storageGroupName) { if (processorMap.containsKey(storageGroupName)) { syncDeleteDataFiles(storageGroupName); } } - private void syncDeleteDataFiles(String storageGroupName) { + private void syncDeleteDataFiles(PartialPath storageGroupName) { logger.info("Force to delete the data in storage group processor {}", storageGroupName); StorageGroupProcessor processor = processorMap.get(storageGroupName); processor.syncDeleteDataFiles(); @@ -517,18 +518,18 @@ public class StorageEngine implements IService { public synchronized boolean deleteAll() { logger.info("Start deleting all storage groups' timeseries"); syncCloseAllProcessor(); - for (String storageGroup : IoTDB.metaManager.getAllStorageGroupNames()) { + for (PartialPath storageGroup : IoTDB.metaManager.getAllStorageGroupNames()) { this.deleteAllDataFilesInOneStorageGroup(storageGroup); } return true; } - public void setTTL(String storageGroup, long dataTTL) throws StorageEngineException { + public void setTTL(PartialPath storageGroup, long dataTTL) throws StorageEngineException { StorageGroupProcessor storageGroupProcessor = getProcessor(storageGroup); storageGroupProcessor.setDataTTL(dataTTL); } - public void deleteStorageGroup(String storageGroupName) { + public void deleteStorageGroup(PartialPath storageGroupName) { deleteAllDataFilesInOneStorageGroup(storageGroupName); StorageGroupProcessor processor = processorMap.remove(storageGroupName); if (processor != null) { @@ -537,8 +538,8 @@ public class StorageEngine implements IService { } public void loadNewTsFileForSync(TsFileResource newTsFileResource) - throws StorageEngineException, LoadFileException { - getProcessor(newTsFileResource.getTsFile().getParentFile().getName()) + throws StorageEngineException, LoadFileException, IllegalPathException { + getProcessor(new PartialPath(newTsFileResource.getTsFile().getParentFile().getName())) .loadNewTsFileForSync(newTsFileResource); } @@ -549,22 +550,24 @@ public class StorageEngine implements IService { throw new StorageEngineException("Can not get the corresponding storage group."); } String device = deviceMap.keySet().iterator().next(); - String storageGroupName = IoTDB.metaManager.getStorageGroupName(device); + PartialPath devicePath = new PartialPath(device); + PartialPath storageGroupName = IoTDB.metaManager.getStorageGroupName(devicePath); getProcessor(storageGroupName).loadNewTsFile(newTsFileResource); } public boolean deleteTsfileForSync(File deletedTsfile) - throws StorageEngineException { - return getProcessor(deletedTsfile.getParentFile().getName()).deleteTsfile(deletedTsfile); + throws StorageEngineException, IllegalPathException { + return getProcessor(new PartialPath(deletedTsfile.getParentFile().getName())).deleteTsfile(deletedTsfile); } - public boolean deleteTsfile(File deletedTsfile) throws StorageEngineException { - return getProcessor(getSgByEngineFile(deletedTsfile)).deleteTsfile(deletedTsfile); + public boolean deleteTsfile(File deletedTsfile) + throws StorageEngineException, IllegalPathException { + return getProcessor(new PartialPath(getSgByEngineFile(deletedTsfile))).deleteTsfile(deletedTsfile); } public boolean moveTsfile(File tsfileToBeMoved, File targetDir) - throws StorageEngineException, IOException { - return getProcessor(getSgByEngineFile(tsfileToBeMoved)).moveTsfile(tsfileToBeMoved, targetDir); + throws StorageEngineException, IOException, IllegalPathException { + return getProcessor(new PartialPath(getSgByEngineFile(tsfileToBeMoved))).moveTsfile(tsfileToBeMoved, targetDir); } /** @@ -581,9 +584,9 @@ public class StorageEngine implements IService { /** * @return TsFiles (seq or unseq) grouped by their storage group and partition number. */ - public Map<String, Map<Long, List<TsFileResource>>> getAllClosedStorageGroupTsFile() { - Map<String, Map<Long, List<TsFileResource>>> ret = new HashMap<>(); - for (Entry<String, StorageGroupProcessor> entry : processorMap.entrySet()) { + public Map<PartialPath, Map<Long, List<TsFileResource>>> getAllClosedStorageGroupTsFile() { + Map<PartialPath, Map<Long, List<TsFileResource>>> ret = new HashMap<>(); + for (Entry<PartialPath, StorageGroupProcessor> entry : processorMap.entrySet()) { List<TsFileResource> allResources = entry.getValue().getSequenceFileTreeSet(); allResources.addAll(entry.getValue().getUnSequenceFileList()); for (TsFileResource sequenceFile : allResources) { @@ -631,13 +634,13 @@ public class StorageEngine implements IService { * @param partitionId * @param newMaxVersion */ - public void setPartitionVersionToMax(String storageGroup, long partitionId, long newMaxVersion) + public void setPartitionVersionToMax(PartialPath storageGroup, long partitionId, long newMaxVersion) throws StorageEngineException { getProcessor(storageGroup).setPartitionFileVersionToMax(partitionId, newMaxVersion); } - public void removePartitions(String storageGroupName, TimePartitionFilter filter) + public void removePartitions(PartialPath storageGroupName, TimePartitionFilter filter) throws StorageEngineException { getProcessor(storageGroupName).removePartitions(filter); } diff --git a/server/src/main/java/org/apache/iotdb/db/engine/modification/Deletion.java b/server/src/main/java/org/apache/iotdb/db/engine/modification/Deletion.java index 879ef73..5421605 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/modification/Deletion.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/modification/Deletion.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.engine.modification; import java.util.Objects; +import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.tsfile.read.common.Path; /** @@ -38,7 +39,7 @@ public class Deletion extends Modification { * @param endTime end time of delete interval * @param path time series path */ - public Deletion(Path path, long versionNum, long endTime) { + public Deletion(PartialPath path, long versionNum, long endTime) { super(Type.DELETION, path, versionNum); this.startTime = Long.MIN_VALUE; this.endTime = endTime; @@ -50,7 +51,7 @@ public class Deletion extends Modification { * @param endTime end time of delete interval * @param path time series path */ - public Deletion(Path path, long versionNum, long startTime, long endTime) { + public Deletion(PartialPath path, long versionNum, long startTime, long endTime) { super(Type.DELETION, path, versionNum); this.startTime = startTime; this.endTime = endTime; diff --git a/server/src/main/java/org/apache/iotdb/db/engine/modification/Modification.java b/server/src/main/java/org/apache/iotdb/db/engine/modification/Modification.java index e75b899..338f8d0 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/modification/Modification.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/modification/Modification.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.engine.modification; import java.util.Objects; +import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.tsfile.read.common.Path; /** @@ -28,32 +29,32 @@ import org.apache.iotdb.tsfile.read.common.Path; public abstract class Modification { protected Type type; - protected Path path; + protected PartialPath path; protected long versionNum; - Modification(Type type, Path path, long versionNum) { + Modification(Type type, PartialPath path, long versionNum) { this.type = type; this.path = path; this.versionNum = versionNum; } public String getPathString() { - return path.getFullPath(); + return path.toString(); } - public Path getPath() { + public PartialPath getPath() { return path; } public String getDevice() { - return path.getDevice(); + return path.getPathWithoutLastNode(); } public String getMeasurement() { - return path.getMeasurement(); + return path.getLastNode(); } - public void setPath(Path path) { + public void setPath(PartialPath path) { this.path = path; } diff --git a/server/src/main/java/org/apache/iotdb/db/engine/querycontext/QueryDataSource.java b/server/src/main/java/org/apache/iotdb/db/engine/querycontext/QueryDataSource.java index 37641c5..1b914bf 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/querycontext/QueryDataSource.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/querycontext/QueryDataSource.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.engine.querycontext; import org.apache.iotdb.db.engine.storagegroup.TsFileResource; +import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.tsfile.read.common.Path; import java.util.List; @@ -28,7 +29,7 @@ import org.apache.iotdb.tsfile.read.filter.basic.Filter; import org.apache.iotdb.tsfile.read.filter.operator.AndFilter; public class QueryDataSource { - private Path seriesPath; + private PartialPath seriesPath; private List<TsFileResource> seqResources; private List<TsFileResource> unseqResources; @@ -37,13 +38,13 @@ public class QueryDataSource { */ private long dataTTL = Long.MAX_VALUE; - public QueryDataSource(Path seriesPath, List<TsFileResource> seqResources, List<TsFileResource> unseqResources) { + public QueryDataSource(PartialPath seriesPath, List<TsFileResource> seqResources, List<TsFileResource> unseqResources) { this.seriesPath = seriesPath; this.seqResources = seqResources; this.unseqResources = unseqResources; } - public Path getSeriesPath() { + public PartialPath getSeriesPath() { return seriesPath; } 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 f384496..a51daea 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 @@ -77,6 +77,7 @@ import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.query.OutOfTTLException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.metadata.MManager; +import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.metadata.mnode.MNode; import org.apache.iotdb.db.metadata.mnode.MeasurementMNode; import org.apache.iotdb.db.qp.physical.crud.DeletePlan; @@ -783,7 +784,7 @@ public class StorageGroupProcessor { // init map long lastFlushTime = partitionLatestFlushedTimeForEachDevice. computeIfAbsent(beforeTimePartition, id -> new HashMap<>()). - computeIfAbsent(insertTabletPlan.getDeviceId(), id -> Long.MIN_VALUE); + computeIfAbsent(insertTabletPlan.getDeviceId().toString(), id -> Long.MIN_VALUE); // if is sequence boolean isSequence = false; while (loc < insertTabletPlan.getRowCount()) { @@ -800,7 +801,7 @@ public class StorageGroupProcessor { beforeTimePartition = curTimePartition; lastFlushTime = partitionLatestFlushedTimeForEachDevice. computeIfAbsent(beforeTimePartition, id -> new HashMap<>()). - computeIfAbsent(insertTabletPlan.getDeviceId(), id -> Long.MIN_VALUE); + computeIfAbsent(insertTabletPlan.getDeviceId().toString(), id -> Long.MIN_VALUE); isSequence = false; } // still in this partition @@ -882,7 +883,7 @@ public class StorageGroupProcessor { .getOrDefault(insertTabletPlan.getDeviceId(), Long.MIN_VALUE) < insertTabletPlan.getTimes()[end - 1]) { latestTimeForEachDevice.get(timePartitionId) - .put(insertTabletPlan.getDeviceId(), insertTabletPlan.getTimes()[end - 1]); + .put(insertTabletPlan.getDeviceId().toString(), insertTabletPlan.getTimes()[end - 1]); } // check memtable size and may async try to flush the work memtable @@ -907,11 +908,13 @@ public class StorageGroupProcessor { if (tmpMeasurementNode != null) { // just for performance, because in single node version, we do not need the full path of measurement // so, we want to avoid concat the device and measurement string in single node version - IoTDB.metaManager.updateLastCache(node.getFullPath(), + IoTDB.metaManager.updateLastCache(node.getPartialPath(), plan.composeLastTimeValuePair(i), true, latestFlushedTime, tmpMeasurementNode); } else { + PartialPath fullPath = node.getPartialPath().clone(); + fullPath.concatPath(new String[]{measurementList[i]}); IoTDB.metaManager - .updateLastCache(node.getFullPath() + IoTDBConstant.PATH_SEPARATOR + measurementList[i], + .updateLastCache(fullPath, plan.composeLastTimeValuePair(i), true, latestFlushedTime, tmpMeasurementNode); } } @@ -932,13 +935,13 @@ public class StorageGroupProcessor { // try to update the latest time of the device of this tsRecord if (latestTimeForEachDevice.get(timePartitionId) - .getOrDefault(insertRowPlan.getDeviceId(), Long.MIN_VALUE) < insertRowPlan.getTime()) { + .getOrDefault(insertRowPlan.getDeviceId().toString(), Long.MIN_VALUE) < insertRowPlan.getTime()) { latestTimeForEachDevice.get(timePartitionId) - .put(insertRowPlan.getDeviceId(), insertRowPlan.getTime()); + .put(insertRowPlan.getDeviceId().toString(), insertRowPlan.getTime()); } long globalLatestFlushTime = globalLatestFlushedTimeForEachDevice.getOrDefault( - insertRowPlan.getDeviceId(), Long.MIN_VALUE); + insertRowPlan.getDeviceId().toString(), Long.MIN_VALUE); tryToUpdateInsertLastCache(insertRowPlan, globalLatestFlushTime); @@ -963,11 +966,14 @@ public class StorageGroupProcessor { if (tmpMeasurementNode != null) { // just for performance, because in single node version, we do not need the full path of measurement // so, we want to avoid concat the device and measurement string in single node version - IoTDB.metaManager.updateLastCache(node.getFullPath(), + IoTDB.metaManager.updateLastCache(node.getPartialPath(), plan.composeTimeValuePair(i), true, latestFlushedTime, tmpMeasurementNode); } else { + PartialPath fullPath; + fullPath = node.getPartialPath().clone(); + fullPath.concatPath(new String[]{measurementList[i]}); IoTDB.metaManager - .updateLastCache(node.getFullPath() + IoTDBConstant.PATH_SEPARATOR + measurementList[i], + .updateLastCache(fullPath, plan.composeTimeValuePair(i), true, latestFlushedTime, tmpMeasurementNode); } } @@ -1334,7 +1340,7 @@ public class StorageGroupProcessor { } // TODO need a read lock, please consider the concurrency with flush manager threads. - public QueryDataSource query(String deviceId, String measurementId, QueryContext context, + public QueryDataSource query(PartialPath deviceId, String measurementId, QueryContext context, QueryFileManager filePathsManager, Filter timeFilter) throws QueryProcessException { insertLock.readLock().lock(); mergeLock.readLock().lock(); @@ -1343,7 +1349,9 @@ public class StorageGroupProcessor { upgradeSeqFileList, deviceId, measurementId, context, timeFilter, true); List<TsFileResource> unseqResources = getFileResourceListForQuery(unSequenceFileList, upgradeUnseqFileList, deviceId, measurementId, context, timeFilter, false); - QueryDataSource dataSource = new QueryDataSource(new Path(deviceId, measurementId), + PartialPath fullPath = deviceId.clone(); + fullPath.concatPath(new String[] {measurementId}); + QueryDataSource dataSource = new QueryDataSource(fullPath, seqResources, unseqResources); // used files should be added before mergeLock is unlocked, or they may be deleted by // running merge @@ -1376,7 +1384,7 @@ public class StorageGroupProcessor { */ private List<TsFileResource> getFileResourceListForQuery( Collection<TsFileResource> tsFileResources, List<TsFileResource> upgradeTsFileResources, - String deviceId, String measurementId, QueryContext context, Filter timeFilter, boolean isSeq) + PartialPath deviceId, String measurementId, QueryContext context, Filter timeFilter, boolean isSeq) throws MetadataException { MeasurementSchema schema = IoTDB.metaManager.getSeriesSchema(deviceId, measurementId); @@ -1387,7 +1395,7 @@ public class StorageGroupProcessor { context.setQueryTimeLowerBound(timeLowerBound); for (TsFileResource tsFileResource : tsFileResources) { - if (!isTsFileResourceSatisfied(tsFileResource, deviceId, timeFilter, isSeq)) { + if (!isTsFileResourceSatisfied(tsFileResource, deviceId.toString(), timeFilter, isSeq)) { continue; } closeQueryLock.readLock().lock(); @@ -1397,7 +1405,7 @@ public class StorageGroupProcessor { } else { tsFileResource.getUnsealedFileProcessor() - .query(deviceId, measurementId, schema.getType(), schema.getEncodingType(), + .query(deviceId.toString(), measurementId, schema.getType(), schema.getEncodingType(), schema.getProps(), context, tsfileResourcesForQuery); } } catch (IOException e) { @@ -1408,7 +1416,7 @@ public class StorageGroupProcessor { } // for upgrade files and old files must be closed for (TsFileResource tsFileResource : upgradeTsFileResources) { - if (!isTsFileResourceSatisfied(tsFileResource, deviceId, timeFilter, isSeq)) { + if (!isTsFileResourceSatisfied(tsFileResource, deviceId.toString(), timeFilter, isSeq)) { continue; } closeQueryLock.readLock().lock(); @@ -1455,7 +1463,7 @@ public class StorageGroupProcessor { * @param startTime the startTime of delete range. * @param endTime the endTime of delete range. */ - public void delete(String deviceId, String measurementId, long startTime, long endTime) + public void delete(PartialPath deviceId, String measurementId, long startTime, long endTime) throws IOException { // TODO: how to avoid partial deletion? // FIXME: notice that if we may remove a SGProcessor out of memory, we need to close all opened @@ -1469,7 +1477,7 @@ public class StorageGroupProcessor { try { Long lastUpdateTime = null; for (Map<String, Long> latestTimeMap : latestTimeForEachDevice.values()) { - Long curTime = latestTimeMap.get(deviceId); + Long curTime = latestTimeMap.get(deviceId.toString()); if (curTime != null && (lastUpdateTime == null || lastUpdateTime < curTime)) { lastUpdateTime = curTime; } @@ -1486,7 +1494,8 @@ public class StorageGroupProcessor { // delete Last cache record if necessary tryToDeleteLastCache(deviceId, measurementId, startTime, endTime); - Path fullPath = new Path(deviceId, measurementId); + PartialPath fullPath = deviceId.clone(); + fullPath.concatPath(new String[]{measurementId}); Deletion deletion = new Deletion(fullPath, MERGE_MOD_START_VERSION_NUM, startTime, endTime); if (mergingModification != null) { mergingModification.write(deletion); @@ -1508,13 +1517,15 @@ public class StorageGroupProcessor { } } - private void logDeletion(long startTime, long endTime, String deviceId, String measurementId) + private void logDeletion(long startTime, long endTime, PartialPath deviceId, String measurementId) throws IOException { long timePartitionStartId = StorageEngine.getTimePartition(startTime); long timePartitionEndId = StorageEngine.getTimePartition(endTime); if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) { + PartialPath fullPath = deviceId.clone(); + fullPath.concatPath(new String[]{measurementId}); DeletePlan deletionPlan = new DeletePlan(startTime, endTime, - new Path(deviceId, measurementId)); + fullPath); for (Map.Entry<Long, TsFileProcessor> entry : workSequenceTsFileProcessors.entrySet()) { if (timePartitionStartId <= entry.getKey() && entry.getKey() <= timePartitionEndId) { entry.getValue().getLogNode().write(deletionPlan); @@ -1560,7 +1571,7 @@ public class StorageGroupProcessor { } } - private void tryToDeleteLastCache(String deviceId, String measurementId, long startTime, + private void tryToDeleteLastCache(PartialPath deviceId, String measurementId, long startTime, long endTime) throws WriteProcessException { MNode node = null; try { 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 75a2ea1..ca40fae 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 @@ -216,11 +216,11 @@ public class TsFileProcessor { } // update start time of this memtable - tsFileResource.updateStartTime(insertRowPlan.getDeviceId(), insertRowPlan.getTime()); + tsFileResource.updateStartTime(insertRowPlan.getDeviceId().toString(), insertRowPlan.getTime()); //for sequence tsfile, we update the endTime only when the file is prepared to be closed. //for unsequence tsfile, we have to update the endTime for each insertion. if (!sequence) { - tsFileResource.updateEndTime(insertRowPlan.getDeviceId(), insertRowPlan.getTime()); + tsFileResource.updateEndTime(insertRowPlan.getDeviceId().toString(), insertRowPlan.getTime()); } } @@ -260,14 +260,14 @@ public class TsFileProcessor { } tsFileResource - .updateStartTime(insertTabletPlan.getDeviceId(), insertTabletPlan.getTimes()[start]); + .updateStartTime(insertTabletPlan.getDeviceId().toString(), insertTabletPlan.getTimes()[start]); //for sequence tsfile, we update the endTime only when the file is prepared to be closed. //for unsequence tsfile, we have to update the endTime for each insertion. if (!sequence) { tsFileResource .updateEndTime( - insertTabletPlan.getDeviceId(), insertTabletPlan.getTimes()[end - 1]); + insertTabletPlan.getDeviceId().toString(), insertTabletPlan.getTimes()[end - 1]); } } diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java index 0650db4..ab7e8f7 100644 --- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java +++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java @@ -27,7 +27,6 @@ import java.io.IOException; import java.nio.file.Files; import java.util.ArrayDeque; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.Comparator; @@ -112,7 +111,7 @@ public class MManager { private TagLogFile tagLogFile; private boolean isRecovering; // device -> DeviceMNode - private RandomDeleteCache<String, MNode> mNodeCache; + private RandomDeleteCache<PartialPath, MNode> mNodeCache; // tag key -> tag value -> LeafMNode private Map<String, Map<String, Set<MeasurementMNode>>> tagIndex = new HashMap<>(); @@ -158,10 +157,10 @@ public class MManager { isRecovering = true; int cacheSize = config.getmManagerCacheSize(); - mNodeCache = new RandomDeleteCache<String, MNode>(cacheSize) { + mNodeCache = new RandomDeleteCache<PartialPath, MNode>(cacheSize) { @Override - public MNode loadObjectByKey(String key) throws CacheException { + public MNode loadObjectByKey(PartialPath key) throws CacheException { lock.readLock().lock(); try { return mtree.getNodeByPathWithStorageGroupCheck(key); @@ -206,10 +205,10 @@ public class MManager { int lineNumber = initFromLog(logFile); if (config.isEnableParameterAdapter()) { - List<String> storageGroups = mtree.getAllStorageGroupNames(); - for (String sg : storageGroups) { - MNode node = mtree.getNodeByPath(new PartialPath(sg)); - seriesNumberInStorageGroups.put(sg, node.getLeafCount()); + List<PartialPath> storageGroups = mtree.getAllStorageGroupNames(); + for (PartialPath sg : storageGroups) { + MNode node = mtree.getNodeByPath(sg); + seriesNumberInStorageGroups.put(sg.toString(), node.getLeafCount()); } maxSeriesNumberAmongStorageGroup = seriesNumberInStorageGroups.values().stream().max(Integer::compareTo).orElse(0); @@ -354,13 +353,13 @@ public class MManager { deleteStorageGroups(storageGroups); break; case MetadataOperationType.SET_TTL: - setTTL(args[1], Long.parseLong(args[2])); + setTTL(new PartialPath(args[1]), Long.parseLong(args[2])); break; case MetadataOperationType.CHANGE_OFFSET: - changeOffset(args[1], Long.parseLong(args[2])); + changeOffset(new PartialPath(args[1]), Long.parseLong(args[2])); break; case MetadataOperationType.CHANGE_ALIAS: - changeAlias(args[1], args[2]); + changeAlias(new PartialPath(args[1]), args[2]); break; default: logger.error("Unrecognizable command {}", cmd); @@ -463,7 +462,7 @@ public class MManager { if (isStorageGroup(prefixPath)) { if (config.isEnableParameterAdapter()) { - int size = seriesNumberInStorageGroups.get(prefixPath); + int size = seriesNumberInStorageGroups.get(prefixPath.toString()); seriesNumberInStorageGroups.put(prefixPath.toString(), 0); if (size == maxSeriesNumberAmongStorageGroup) { seriesNumberInStorageGroups.values().stream() @@ -482,7 +481,7 @@ public class MManager { Set<String> failedNames = new HashSet<>(); for (PartialPath p : allTimeseries) { try { - String emptyStorageGroup = deleteOneTimeseriesAndUpdateStatistics(p); + PartialPath emptyStorageGroup = deleteOneTimeseriesAndUpdateStatistics(p); if (!isRecovering) { if (emptyStorageGroup != null) { StorageEngine.getInstance().deleteAllDataFilesInOneStorageGroup(emptyStorageGroup); @@ -544,13 +543,13 @@ public class MManager { * @param path full path from root to leaf node * @return after delete if the storage group is empty, return its name, otherwise return null */ - private String deleteOneTimeseriesAndUpdateStatistics(PartialPath path) + private PartialPath deleteOneTimeseriesAndUpdateStatistics(PartialPath path) throws MetadataException, IOException { lock.writeLock().lock(); try { - Pair<String, MeasurementMNode> pair = mtree.deleteTimeseriesAndReturnEmptyStorageGroup(path); + Pair<PartialPath, MeasurementMNode> pair = mtree.deleteTimeseriesAndReturnEmptyStorageGroup(path); removeFromTagInvertedIndex(pair.right); - String storageGroupName = pair.left; + PartialPath storageGroupName = pair.left; // TODO: delete the path node and all its ancestors mNodeCache.clear(); @@ -561,9 +560,9 @@ public class MManager { } if (config.isEnableParameterAdapter()) { - String storageGroup = getStorageGroupName(path).toString(); - int size = seriesNumberInStorageGroups.get(storageGroup); - seriesNumberInStorageGroups.put(storageGroup, size - 1); + PartialPath storageGroup = getStorageGroupName(path); + int size = seriesNumberInStorageGroups.get(storageGroup.toString()); + seriesNumberInStorageGroups.put(storageGroup.toString(), size - 1); if (size == maxSeriesNumberAmongStorageGroup) { seriesNumberInStorageGroups.values().stream().max(Integer::compareTo) .ifPresent(val -> maxSeriesNumberAmongStorageGroup = val); @@ -680,7 +679,7 @@ public class MManager { } } - public MeasurementSchema[] getSchemas(String deviceId, String[] measurements) + public MeasurementSchema[] getSchemas(PartialPath deviceId, String[] measurements) throws MetadataException { lock.readLock().lock(); try { @@ -757,7 +756,7 @@ public class MManager { /** * Get all storage group names */ - public List<String> getAllStorageGroupNames() { + public List<PartialPath> getAllStorageGroupNames() { lock.readLock().lock(); try { return mtree.getAllStorageGroupNames(); @@ -954,24 +953,24 @@ public class MManager { ans = mtree.getAllMeasurementSchema(plan); } List<ShowTimeSeriesResult> res = new LinkedList<>(); - for (String[] ansString : ans) { - long tagFileOffset = Long.parseLong(ansString[6]); + for (Pair<PartialPath, String[]> ansString : ans) { + long tagFileOffset = Long.parseLong(ansString.right[5]); try { if (tagFileOffset < 0) { // no tags/attributes - res.add(new ShowTimeSeriesResult(ansString[0], ansString[1], ansString[2], ansString[3], - ansString[4], ansString[5], Collections.emptyMap())); + res.add(new ShowTimeSeriesResult(ansString.left.toString(), ansString.right[0], ansString.right[1], ansString.right[2], + ansString.right[3], ansString.right[4], Collections.emptyMap())); } else { // has tags/attributes Pair<Map<String, String>, Map<String, String>> pair = tagLogFile.read(config.getTagAttributeTotalSize(), tagFileOffset); pair.left.putAll(pair.right); - res.add(new ShowTimeSeriesResult(ansString[0], ansString[1], ansString[2], ansString[3], - ansString[4], ansString[5], pair.left)); + res.add(new ShowTimeSeriesResult(ansString.left.toString(), ansString.right[0], ansString.right[1], ansString.right[2], + ansString.right[3], ansString.right[4], pair.left)); } } catch (IOException e) { throw new MetadataException( - "Something went wrong while deserialize tag info of " + ansString[0], e); + "Something went wrong while deserialize tag info of " + ansString.left.toString(), e); } } return res; @@ -980,7 +979,7 @@ public class MManager { } } - public MeasurementSchema getSeriesSchema(String device, String measurement) + public MeasurementSchema getSeriesSchema(PartialPath device, String measurement) throws MetadataException { lock.readLock().lock(); try { @@ -1020,7 +1019,7 @@ public class MManager { * * @param path a full path or a prefix path */ - public boolean isPathExist(String path) { + public boolean isPathExist(PartialPath path) { lock.readLock().lock(); try { return mtree.isPathExist(path); @@ -1032,7 +1031,7 @@ public class MManager { /** * Get node by path */ - public MNode getNodeByPath(String path) throws MetadataException { + public MNode getNodeByPath(PartialPath path) throws MetadataException { lock.readLock().lock(); try { return mtree.getNodeByPath(path); @@ -1045,7 +1044,7 @@ public class MManager { * Get storage group node by path. If storage group is not set, StorageGroupNotSetException will * be thrown */ - public StorageGroupMNode getStorageGroupNode(String path) throws MetadataException { + public StorageGroupMNode getStorageGroupNode(PartialPath path) throws MetadataException { lock.readLock().lock(); try { return mtree.getStorageGroupNode(path); @@ -1063,7 +1062,7 @@ public class MManager { * @param path path */ public MNode getDeviceNodeWithAutoCreateAndReadLock( - String path, boolean autoCreateSchema, int sgLevel) throws MetadataException { + PartialPath path, boolean autoCreateSchema, int sgLevel) throws MetadataException { lock.readLock().lock(); MNode node = null; boolean shouldSetStorageGroup; @@ -1072,7 +1071,7 @@ public class MManager { return node; } catch (CacheException e) { if (!autoCreateSchema) { - throw new PathNotExistException(path); + throw new PathNotExistException(path.toString()); } } finally { if (node != null) { @@ -1091,7 +1090,7 @@ public class MManager { } if (shouldSetStorageGroup) { - String storageGroupName = MetaUtils.getStorageGroupNameByLevel(path, sgLevel); + PartialPath storageGroupName = MetaUtils.getStorageGroupNameByLevel(path, sgLevel); setStorageGroup(storageGroupName); } node = mtree.getDeviceNodeWithAutoCreating(path, sgLevel); @@ -1111,19 +1110,19 @@ public class MManager { /** * !!!!!!Attention!!!!! must call the return node's readUnlock() if you call this method. */ - public MNode getDeviceNodeWithAutoCreateAndReadLock(String path) throws MetadataException { + public MNode getDeviceNodeWithAutoCreateAndReadLock(PartialPath path) throws MetadataException { return getDeviceNodeWithAutoCreateAndReadLock( path, config.isAutoCreateSchemaEnabled(), config.getDefaultStorageGroupLevel()); } - public MNode getDeviceNode(String path) throws MetadataException { + public MNode getDeviceNode(PartialPath path) throws MetadataException { lock.readLock().lock(); MNode node; try { node = mNodeCache.get(path); return node; } catch (CacheException e) { - throw new PathNotExistException(path); + throw new PathNotExistException(path.toString()); } finally { lock.readLock().unlock(); } @@ -1136,15 +1135,16 @@ public class MManager { * @param path read from disk * @return deviceId */ - public String getDeviceId(String path) { + public String getDeviceId(PartialPath path) { MNode deviceNode = null; + String device = null; try { deviceNode = getDeviceNode(path); - path = deviceNode.getFullPath(); + device = deviceNode.getFullPath(); } catch (MetadataException | NullPointerException e) { // Cannot get deviceId from MManager, return the input deviceId } - return path; + return device; } public MNode getChild(MNode parent, String child) { @@ -1177,12 +1177,12 @@ public class MManager { return maxSeriesNumberAmongStorageGroup; } - public void setTTL(String storageGroup, long dataTTL) throws MetadataException, IOException { + public void setTTL(PartialPath storageGroup, long dataTTL) throws MetadataException, IOException { lock.writeLock().lock(); try { getStorageGroupNode(storageGroup).setDataTTL(dataTTL); if (!isRecovering) { - logWriter.setTTL(storageGroup, dataTTL); + logWriter.setTTL(storageGroup.toString(), dataTTL); } } finally { lock.writeLock().unlock(); @@ -1194,11 +1194,11 @@ public class MManager { * * @return key-> storageGroupName, value->ttl */ - public Map<String, Long> getStorageGroupsTTL() { - Map<String, Long> storageGroupsTTL = new HashMap<>(); + public Map<PartialPath, Long> getStorageGroupsTTL() { + Map<PartialPath, Long> storageGroupsTTL = new HashMap<>(); try { - List<String> storageGroups = this.getAllStorageGroupNames(); - for (String storageGroup : storageGroups) { + List<PartialPath> storageGroups = this.getAllStorageGroupNames(); + for (PartialPath storageGroup : storageGroups) { long ttl = getStorageGroupNode(storageGroup).getDataTTL(); storageGroupsTTL.put(storageGroup, ttl); } @@ -1215,7 +1215,7 @@ public class MManager { * @param path timeseries * @param offset offset in the tag file */ - public void changeOffset(String path, long offset) throws MetadataException { + public void changeOffset(PartialPath path, long offset) throws MetadataException { lock.writeLock().lock(); try { ((MeasurementMNode) mtree.getNodeByPath(path)).setOffset(offset); @@ -1224,7 +1224,7 @@ public class MManager { } } - public void changeAlias(String path, String alias) throws MetadataException { + public void changeAlias(PartialPath path, String alias) throws MetadataException { lock.writeLock().lock(); try { MeasurementMNode leafMNode = (MeasurementMNode) mtree.getNodeByPath(path); @@ -1248,12 +1248,12 @@ public class MManager { * @param fullPath timeseries */ public void upsertTagsAndAttributes(String alias, Map<String, String> tagsMap, - Map<String, String> attributesMap, String fullPath) throws MetadataException, IOException { + Map<String, String> attributesMap, PartialPath fullPath) throws MetadataException, IOException { lock.writeLock().lock(); try { MNode mNode = mtree.getNodeByPath(fullPath); if (!(mNode instanceof MeasurementMNode)) { - throw new PathNotExistException(fullPath); + throw new PathNotExistException(fullPath.toString()); } MeasurementMNode leafMNode = (MeasurementMNode) mNode; // upsert alias @@ -1268,7 +1268,7 @@ public class MManager { leafMNode.getParent().addAlias(alias, leafMNode); leafMNode.setAlias(alias); // persist to WAL - logWriter.changeAlias(fullPath, alias); + logWriter.changeAlias(fullPath.toString(), alias); } if (tagsMap == null && attributesMap == null) { @@ -1277,7 +1277,7 @@ public class MManager { // no tag or attribute, we need to add a new record in log if (leafMNode.getOffset() < 0) { long offset = tagLogFile.write(tagsMap, attributesMap); - logWriter.changeOffset(fullPath, offset); + logWriter.changeOffset(fullPath.toString(), offset); leafMNode.setOffset(offset); // update inverted Index map if (tagsMap != null) { @@ -1350,19 +1350,19 @@ public class MManager { * @param attributesMap newly added attributes map * @param fullPath timeseries */ - public void addAttributes(Map<String, String> attributesMap, String fullPath) + public void addAttributes(Map<String, String> attributesMap, PartialPath fullPath) throws MetadataException, IOException { lock.writeLock().lock(); try { MNode mNode = mtree.getNodeByPath(fullPath); if (!(mNode instanceof MeasurementMNode)) { - throw new PathNotExistException(fullPath); + throw new PathNotExistException(fullPath.toString()); } MeasurementMNode leafMNode = (MeasurementMNode) mNode; // no tag or attribute, we need to add a new record in log if (leafMNode.getOffset() < 0) { long offset = tagLogFile.write(Collections.emptyMap(), attributesMap); - logWriter.changeOffset(fullPath, offset); + logWriter.changeOffset(fullPath.toString(), offset); leafMNode.setOffset(offset); return; } @@ -1393,19 +1393,19 @@ public class MManager { * @param tagsMap newly added tags map * @param fullPath timeseries */ - public void addTags(Map<String, String> tagsMap, String fullPath) + public void addTags(Map<String, String> tagsMap, PartialPath fullPath) throws MetadataException, IOException { lock.writeLock().lock(); try { MNode mNode = mtree.getNodeByPath(fullPath); if (!(mNode instanceof MeasurementMNode)) { - throw new PathNotExistException(fullPath); + throw new PathNotExistException(fullPath.toString()); } MeasurementMNode leafMNode = (MeasurementMNode) mNode; // no tag or attribute, we need to add a new record in log if (leafMNode.getOffset() < 0) { long offset = tagLogFile.write(tagsMap, Collections.emptyMap()); - logWriter.changeOffset(fullPath, offset); + logWriter.changeOffset(fullPath.toString(), offset); leafMNode.setOffset(offset); // update inverted Index map for (Entry<String, String> entry : tagsMap.entrySet()) { @@ -1446,13 +1446,13 @@ public class MManager { * @param keySet tags key or attributes key * @param fullPath timeseries path */ - public void dropTagsOrAttributes(Set<String> keySet, String fullPath) + public void dropTagsOrAttributes(Set<String> keySet, PartialPath fullPath) throws MetadataException, IOException { lock.writeLock().lock(); try { MNode mNode = mtree.getNodeByPath(fullPath); if (!(mNode instanceof MeasurementMNode)) { - throw new PathNotExistException(fullPath); + throw new PathNotExistException(fullPath.toString()); } MeasurementMNode leafMNode = (MeasurementMNode) mNode; // no tag or attribute, just do nothing. @@ -1517,13 +1517,13 @@ public class MManager { * @param alterMap the new tags or attributes key-value * @param fullPath timeseries */ - public void setTagsOrAttributesValue(Map<String, String> alterMap, String fullPath) + public void setTagsOrAttributesValue(Map<String, String> alterMap, PartialPath fullPath) throws MetadataException, IOException { lock.writeLock().lock(); try { MNode mNode = mtree.getNodeByPath(fullPath); if (!(mNode instanceof MeasurementMNode)) { - throw new PathNotExistException(fullPath); + throw new PathNotExistException(fullPath.toString()); } MeasurementMNode leafMNode = (MeasurementMNode) mNode; if (leafMNode.getOffset() < 0) { @@ -1596,13 +1596,13 @@ public class MManager { * @param newKey new key of tag or attribute * @param fullPath timeseries */ - public void renameTagOrAttributeKey(String oldKey, String newKey, String fullPath) + public void renameTagOrAttributeKey(String oldKey, String newKey, PartialPath fullPath) throws MetadataException, IOException { lock.writeLock().lock(); try { MNode mNode = mtree.getNodeByPath(fullPath); if (!(mNode instanceof MeasurementMNode)) { - throw new PathNotExistException(fullPath); + throw new PathNotExistException(fullPath.toString()); } MeasurementMNode leafMNode = (MeasurementMNode) mNode; if (leafMNode.getOffset() < 0) { @@ -1666,7 +1666,7 @@ public class MManager { /** * Check whether the given path contains a storage group */ - boolean checkStorageGroupByPath(String path) { + boolean checkStorageGroupByPath(PartialPath path) { lock.readLock().lock(); try { return mtree.checkStorageGroupByPath(path); @@ -1681,7 +1681,7 @@ public class MManager { * @return List of String represented all storage group names * @apiNote :for cluster */ - List<String> getStorageGroupByPath(String path) throws MetadataException { + List<String> getStorageGroupByPath(PartialPath path) throws MetadataException { lock.readLock().lock(); try { return mtree.getStorageGroupByPath(path); @@ -1730,7 +1730,7 @@ public class MManager { * @param startingPath * @param measurementSchemas */ - public void collectSeries(String startingPath, List<MeasurementSchema> measurementSchemas) { + public void collectSeries(PartialPath startingPath, List<MeasurementSchema> measurementSchemas) { MNode mNode; try { mNode = getNodeByPath(startingPath); @@ -1791,7 +1791,7 @@ public class MManager { // do nothing } - public void updateLastCache(String seriesPath, TimeValuePair timeValuePair, + public void updateLastCache(PartialPath seriesPath, TimeValuePair timeValuePair, boolean highPriorityUpdate, Long latestFlushedTime, MeasurementMNode node) { if (node != null) { @@ -1806,7 +1806,7 @@ public class MManager { } } - public TimeValuePair getLastCache(String seriesPath) { + public TimeValuePair getLastCache(PartialPath seriesPath) { try { MeasurementMNode node = (MeasurementMNode) mtree.getNodeByPath(seriesPath); return node.getCachedLast(); @@ -1873,7 +1873,7 @@ public class MManager { * * @throws MetadataException */ - public MeasurementSchema[] getSeriesSchemasAndReadLockDevice(String deviceId, + public MeasurementSchema[] getSeriesSchemasAndReadLockDevice(PartialPath deviceId, String[] measurementList, InsertPlan plan) throws MetadataException { MeasurementSchema[] schemas = new MeasurementSchema[measurementList.length]; @@ -1894,11 +1894,12 @@ public class MManager { } // create it - Path path = new Path(deviceId, measurementList[i]); - TSDataType dataType = getTypeInLoc(plan, i); + TSDataType dataType = getTypeInLoc(plan, i); + PartialPath path = deviceId.clone(); + path.concatPath(new String[]{measurementList[i]}); createTimeseries( - path.getFullPath(), + path, dataType, getDefaultEncoding(dataType), TSFileDescriptor.getInstance().getConfig().getCompressor(), @@ -2006,7 +2007,7 @@ public class MManager { * * @param deviceId */ - public void unlockDeviceReadLock(String deviceId) { + public void unlockDeviceReadLock(PartialPath deviceId) { try { MNode mNode = getDeviceNode(deviceId); mNode.readUnlock(); diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java b/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java index 6843d4e..41c3316 100644 --- a/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java +++ b/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java @@ -161,10 +161,10 @@ public class MTree implements Serializable { * * <p>e.g., get root.sg.d1, get or create all internal nodes and return the node of d1 */ - MNode getDeviceNodeWithAutoCreating(String deviceId, int sgLevel) throws MetadataException { - String[] nodeNames = MetaUtils.getNodeNames(deviceId); + MNode getDeviceNodeWithAutoCreating(PartialPath deviceId, int sgLevel) throws MetadataException { + String[] nodeNames = deviceId.getNodes(); if (nodeNames.length <= 1 || !nodeNames[0].equals(root.getName())) { - throw new IllegalPathException(deviceId); + throw new IllegalPathException(deviceId.toString()); } MNode cur = root; for (int i = 1; i < nodeNames.length; i++) { @@ -186,8 +186,8 @@ public class MTree implements Serializable { * * @param path a full path or a prefix path */ - boolean isPathExist(String path) { - String[] nodeNames = MetaUtils.getNodeNames(path); + boolean isPathExist(PartialPath path) { + String[] nodeNames = path.getNodes(); MNode cur = root; if (!nodeNames[0].equals(root.getName())) { return false; @@ -306,13 +306,13 @@ public class MTree implements Serializable { * * @param path Format: root.node(.node)+ */ - Pair<String, MeasurementMNode> deleteTimeseriesAndReturnEmptyStorageGroup(PartialPath path) + Pair<PartialPath, MeasurementMNode> deleteTimeseriesAndReturnEmptyStorageGroup(PartialPath path) throws MetadataException { MNode curNode = getNodeByPath(path); if (!(curNode instanceof MeasurementMNode)) { throw new PathNotExistException(path.toString()); } - String[] nodes = MetaUtils.getNodeNames(path.toString()); + String[] nodes = path.getNodes(); if (nodes.length == 0 || !IoTDBConstant.PATH_ROOT.equals(nodes[0])) { throw new IllegalPathException(path.toString()); } @@ -328,7 +328,7 @@ public class MTree implements Serializable { && curNode.getChildren().size() == 0) { // if current storage group has no time series, return the storage group name if (curNode instanceof StorageGroupMNode) { - return new Pair<>(curNode.getFullPath(), deletedNode); + return new Pair<>(curNode.getPartialPath(), deletedNode); } curNode.getParent().deleteChild(curNode.getName()); curNode = curNode.getParent(); @@ -348,20 +348,20 @@ public class MTree implements Serializable { * Get node by path with storage group check If storage group is not set, * StorageGroupNotSetException will be thrown */ - MNode getNodeByPathWithStorageGroupCheck(String path) throws MetadataException { + MNode getNodeByPathWithStorageGroupCheck(PartialPath path) throws MetadataException { boolean storageGroupChecked = false; - String[] nodes = MetaUtils.getNodeNames(path); + String[] nodes = path.getNodes(); if (nodes.length == 0 || !nodes[0].equals(root.getName())) { - throw new IllegalPathException(path); + throw new IllegalPathException(path.toString()); } MNode cur = root; for (int i = 1; i < nodes.length; i++) { if (!cur.hasChild(nodes[i])) { if (!storageGroupChecked) { - throw new StorageGroupNotSetException(path); + throw new StorageGroupNotSetException(path.toString()); } - throw new PathNotExistException(path); + throw new PathNotExistException(path.toString()); } cur = cur.getChild(nodes[i]); @@ -371,7 +371,7 @@ public class MTree implements Serializable { } if (!storageGroupChecked) { - throw new StorageGroupNotSetException(path); + throw new StorageGroupNotSetException(path.toString()); } return cur; } @@ -414,11 +414,11 @@ public class MTree implements Serializable { * @return storage group list * @apiNote :for cluster */ - List<String> getStorageGroupByPath(String path) throws MetadataException { + List<String> getStorageGroupByPath(PartialPath path) throws MetadataException { List<String> storageGroups = new ArrayList<>(); - String[] nodes = MetaUtils.getNodeNames(path); + String[] nodes = path.getNodes(); if (nodes.length == 0 || !nodes[0].equals(root.getName())) { - throw new IllegalPathException(path); + throw new IllegalPathException(path.toString()); } findStorageGroup(root, nodes, 1, "", storageGroups); return storageGroups; @@ -458,14 +458,14 @@ public class MTree implements Serializable { * * @return a list contains all distinct storage groups */ - List<String> getAllStorageGroupNames() { - List<String> res = new ArrayList<>(); + List<PartialPath> getAllStorageGroupNames() { + List<PartialPath> res = new ArrayList<>(); Deque<MNode> nodeStack = new ArrayDeque<>(); nodeStack.add(root); while (!nodeStack.isEmpty()) { MNode current = nodeStack.pop(); if (current instanceof StorageGroupMNode) { - res.add(current.getFullPath()); + res.add(current.getPartialPath()); } else { nodeStack.addAll(current.getChildren().values()); } @@ -515,8 +515,8 @@ public class MTree implements Serializable { /** * Check whether the given path contains a storage group */ - boolean checkStorageGroupByPath(String path) { - String[] nodes = MetaUtils.getNodeNames(path); + boolean checkStorageGroupByPath(PartialPath path) { + String[] nodes = path.getNodes(); MNode cur = root; for (int i = 1; i <= nodes.length; i++) { cur = cur.getChild(nodes[i]); diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java index 7f7bfc1..6e7cfe3 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java @@ -25,6 +25,7 @@ import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException; import org.apache.iotdb.db.exception.query.QueryProcessException; +import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.qp.physical.PhysicalPlan; import org.apache.iotdb.db.qp.physical.crud.DeletePlan; import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan; @@ -81,7 +82,7 @@ public interface IPlanExecutor { * @param startTime start time in delete command * @param endTime end time in delete command */ - void delete(Path path, long startTime, long endTime) throws QueryProcessException; + void delete(PartialPath path, long startTime, long endTime) throws QueryProcessException; /** * execute insert command and return whether the operator is successful. 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 43b0c2d..9a70bf0 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 @@ -70,6 +70,7 @@ import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartiti import org.apache.iotdb.db.engine.storagegroup.TsFileResource; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.exception.metadata.DeleteFailedException; +import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException; import org.apache.iotdb.db.exception.query.QueryProcessException; @@ -78,6 +79,7 @@ import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.metadata.mnode.MNode; import org.apache.iotdb.db.metadata.mnode.MeasurementMNode; import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode; +import org.apache.iotdb.db.qp.constant.SQLConstant; import org.apache.iotdb.db.qp.logical.Operator.OperatorType; import org.apache.iotdb.db.qp.logical.sys.AuthorOperator; import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType; @@ -128,6 +130,7 @@ import org.apache.iotdb.db.utils.AuthUtils; import org.apache.iotdb.db.utils.FileLoaderUtils; import org.apache.iotdb.db.utils.QueryUtils; import org.apache.iotdb.db.utils.UpgradeUtils; +import org.apache.iotdb.tsfile.common.constant.TsFileConstant; import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException; import org.apache.iotdb.tsfile.file.metadata.ChunkGroupMetadata; import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata; @@ -258,7 +261,7 @@ public class PlanExecutor implements IPlanExecutor { DeletePartitionPlan p = (DeletePartitionPlan) plan; TimePartitionFilter filter = (storageGroupName, partitionId) -> - storageGroupName.equals(((DeletePartitionPlan) plan).getStorageGroupName()) + storageGroupName.equals(((DeletePartitionPlan) plan).getStorageGroupName().toString()) && p.getPartitionId().contains(partitionId); StorageEngine.getInstance() .removePartitions(((DeletePartitionPlan) plan).getStorageGroupName(), filter); @@ -425,7 +428,7 @@ public class PlanExecutor implements IPlanExecutor { return IoTDB.metaManager.getNodesCountInGivenLevel(path, level); } - protected List<String> getPathsName(String path) throws MetadataException { + protected List<PartialPath> getPathsName(PartialPath path) throws MetadataException { return IoTDB.metaManager.getAllTimeseriesName(path); } @@ -453,18 +456,18 @@ public class PlanExecutor implements IPlanExecutor { new ListDataSet( Collections.singletonList(new Path(COLUMN_DEVICES)), Collections.singletonList(TSDataType.TEXT)); - Set<String> devices = getDevices(showDevicesPlan.getPath().toString()); - for (String s : devices) { + Set<PartialPath> devices = getDevices(showDevicesPlan.getPath()); + for (PartialPath s : devices) { RowRecord record = new RowRecord(0); Field field = new Field(TSDataType.TEXT); - field.setBinaryV(new Binary(s)); + field.setBinaryV(new Binary(s.toString())); record.addField(field); listDataSet.putRecord(record); } return listDataSet; } - protected Set<String> getDevices(String path) throws MetadataException { + protected Set<PartialPath> getDevices(PartialPath path) throws MetadataException { return IoTDB.metaManager.getDevices(path); } @@ -489,7 +492,7 @@ public class PlanExecutor implements IPlanExecutor { return IoTDB.metaManager.getChildNodePathInNextLevel(path); } - protected List<String> getAllStorageGroupNames() { + protected List<PartialPath> getAllStorageGroupNames() { return IoTDB.metaManager.getAllStorageGroupNames(); } @@ -498,11 +501,11 @@ public class PlanExecutor implements IPlanExecutor { new ListDataSet( Collections.singletonList(new Path(COLUMN_STORAGE_GROUP)), Collections.singletonList(TSDataType.TEXT)); - List<String> storageGroupList = getAllStorageGroupNames(); - for (String s : storageGroupList) { + List<PartialPath> storageGroupList = getAllStorageGroupNames(); + for (PartialPath s : storageGroupList) { RowRecord record = new RowRecord(0); Field field = new Field(TSDataType.TEXT); - field.setBinaryV(new Binary(s)); + field.setBinaryV(new Binary(s.toString())); record.addField(field); listDataSet.putRecord(record); } @@ -529,19 +532,19 @@ public class PlanExecutor implements IPlanExecutor { new ListDataSet( Arrays.asList(new Path(COLUMN_STORAGE_GROUP), new Path(COLUMN_TTL)), Arrays.asList(TSDataType.TEXT, TSDataType.INT64)); - List<String> selectedSgs = showTTLPlan.getStorageGroups(); + List<PartialPath> selectedSgs = showTTLPlan.getStorageGroups(); List<StorageGroupMNode> storageGroups = getAllStorageGroupNodes(); int timestamp = 0; for (StorageGroupMNode mNode : storageGroups) { - String sgName = mNode.getFullPath(); + PartialPath sgName = mNode.getPartialPath(); if (!selectedSgs.isEmpty() && !selectedSgs.contains(sgName)) { continue; } RowRecord rowRecord = new RowRecord(timestamp++); Field sg = new Field(TSDataType.TEXT); Field ttl; - sg.setBinaryV(new Binary(sgName)); + sg.setBinaryV(new Binary(sgName.toString())); if (mNode.getDataTTL() != Long.MAX_VALUE) { ttl = new Field(TSDataType.INT64); ttl.setLongV(mNode.getDataTTL()); @@ -654,15 +657,15 @@ public class PlanExecutor implements IPlanExecutor { @Override public void delete(DeletePlan deletePlan) throws QueryProcessException { try { - Set<String> existingPaths = new HashSet<>(); - for (Path p : deletePlan.getPaths()) { - existingPaths.addAll(getPathsName(p.getFullPath())); + Set<PartialPath> existingPaths = new HashSet<>(); + for (PartialPath p : deletePlan.getPaths()) { + existingPaths.addAll(getPathsName(p)); } if (existingPaths.isEmpty()) { throw new QueryProcessException("TimeSeries does not exist and its data cannot be deleted"); } - for (String path : existingPaths) { - delete(new Path(path), deletePlan.getDeleteStartTime(), deletePlan.getDeleteEndTime()); + for (PartialPath path : existingPaths) { + delete(path, deletePlan.getDeleteStartTime(), deletePlan.getDeleteEndTime()); } } catch (MetadataException e) { throw new QueryProcessException(e); @@ -750,17 +753,17 @@ public class PlanExecutor implements IPlanExecutor { return; } - Set<Path> registeredSeries = new HashSet<>(); + Set<PartialPath> registeredSeries = new HashSet<>(); for (ChunkGroupMetadata chunkGroupMetadata : chunkGroupMetadataList) { String device = chunkGroupMetadata.getDevice(); MNode node = null; try { - node = mManager.getDeviceNodeWithAutoCreateAndReadLock(device, true, sgLevel); + node = mManager.getDeviceNodeWithAutoCreateAndReadLock(new PartialPath(device), true, sgLevel); for (ChunkMetadata chunkMetadata : chunkGroupMetadata.getChunkMetadataList()) { - Path series = new Path(chunkGroupMetadata.getDevice(), chunkMetadata.getMeasurementUid()); + PartialPath series = new PartialPath(chunkGroupMetadata.getDevice() + TsFileConstant.PATH_SEPARATOR + chunkMetadata.getMeasurementUid()); if (!registeredSeries.contains(series)) { registeredSeries.add(series); - MeasurementSchema schema = knownSchemas.get(series); + MeasurementSchema schema = knownSchemas.get(new Path(series.getPathWithoutLastNode(), series.getLastNode())); if (schema == null) { throw new MetadataException( String.format( @@ -769,7 +772,7 @@ public class PlanExecutor implements IPlanExecutor { } if (!node.hasChild(chunkMetadata.getMeasurementUid())) { mManager.createTimeseries( - series.getFullPath(), + series, schema.getType(), schema.getEncodingType(), schema.getCompressor(), @@ -795,7 +798,7 @@ public class PlanExecutor implements IPlanExecutor { throw new QueryProcessException( String.format("File %s doesn't exist.", plan.getFile().getName())); } - } catch (StorageEngineException e) { + } catch (StorageEngineException | IllegalPathException e) { throw new QueryProcessException( String.format("Cannot remove file because %s", e.getMessage())); } @@ -811,7 +814,7 @@ public class PlanExecutor implements IPlanExecutor { throw new QueryProcessException( String.format("File %s doesn't exist.", plan.getFile().getName())); } - } catch (StorageEngineException | IOException e) { + } catch (StorageEngineException | IOException | IllegalPathException e) { throw new QueryProcessException( String.format( "Cannot move file %s to target directory %s because %s", @@ -836,17 +839,16 @@ public class PlanExecutor implements IPlanExecutor { } @Override - public void delete(Path path, long startTime, long endTime) throws QueryProcessException { - String deviceId = path.getDevice(); - String measurementId = path.getMeasurement(); + public void delete(PartialPath path, long startTime, long endTime) throws QueryProcessException { + PartialPath deviceId = new PartialPath(Arrays.copyOf(path.getNodes(), path.getNodes().length-1)); + String measurementId = path.getLastNode(); try { - if (!mManager.isPathExist(path.getFullPath())) { + if (!mManager.isPathExist(path)) { throw new QueryProcessException( - String.format("Time series %s does not exist.", path.getFullPath())); + String.format("Time series %s does not exist.", path.toString())); } - mManager.getStorageGroupName(path.getFullPath()); StorageEngine.getInstance().delete(deviceId, measurementId, startTime, endTime); - } catch (MetadataException | StorageEngineException e) { + } catch (StorageEngineException e) { throw new QueryProcessException(e); } } @@ -899,7 +901,7 @@ public class PlanExecutor implements IPlanExecutor { String password = author.getPassword(); String newPassword = author.getNewPassword(); Set<Integer> permissions = author.getPermissions(); - Path nodeName = author.getNodeName(); + PartialPath nodeName = author.getNodeName(); try { switch (authorType) { case UPDATE_USER: @@ -919,12 +921,12 @@ public class PlanExecutor implements IPlanExecutor { break; case GRANT_ROLE: for (int i : permissions) { - authorizer.grantPrivilegeToRole(roleName, nodeName.getFullPath(), i); + authorizer.grantPrivilegeToRole(roleName, nodeName.toString(), i); } break; case GRANT_USER: for (int i : permissions) { - authorizer.grantPrivilegeToUser(userName, nodeName.getFullPath(), i); + authorizer.grantPrivilegeToUser(userName, nodeName.toString(), i); } break; case GRANT_ROLE_TO_USER: @@ -932,12 +934,12 @@ public class PlanExecutor implements IPlanExecutor { break; case REVOKE_USER: for (int i : permissions) { - authorizer.revokePrivilegeFromUser(userName, nodeName.getFullPath(), i); + authorizer.revokePrivilegeFromUser(userName, nodeName.toString(), i); } break; case REVOKE_ROLE: for (int i : permissions) { - authorizer.revokePrivilegeFromRole(roleName, nodeName.getFullPath(), i); + authorizer.revokePrivilegeFromRole(roleName, nodeName.toString(), i); } break; case REVOKE_ROLE_FROM_USER: @@ -976,12 +978,12 @@ public class PlanExecutor implements IPlanExecutor { protected boolean deleteTimeSeries(DeleteTimeSeriesPlan deleteTimeSeriesPlan) throws QueryProcessException { - List<Path> deletePathList = deleteTimeSeriesPlan.getPaths(); + List<PartialPath> deletePathList = deleteTimeSeriesPlan.getPaths(); try { deleteDataOfTimeSeries(deletePathList); List<String> failedNames = new LinkedList<>(); - for (Path path : deletePathList) { - String failedTimeseries = mManager.deleteTimeseries(path.toString()); + for (PartialPath path : deletePathList) { + String failedTimeseries = mManager.deleteTimeseries(path); if (!failedTimeseries.isEmpty()) { failedNames.add(failedTimeseries); } @@ -997,31 +999,31 @@ public class PlanExecutor implements IPlanExecutor { private boolean alterTimeSeries(AlterTimeSeriesPlan alterTimeSeriesPlan) throws QueryProcessException { - Path path = alterTimeSeriesPlan.getPath(); + PartialPath path = alterTimeSeriesPlan.getPath(); Map<String, String> alterMap = alterTimeSeriesPlan.getAlterMap(); try { switch (alterTimeSeriesPlan.getAlterType()) { case RENAME: String beforeName = alterMap.keySet().iterator().next(); String currentName = alterMap.get(beforeName); - mManager.renameTagOrAttributeKey(beforeName, currentName, path.getFullPath()); + mManager.renameTagOrAttributeKey(beforeName, currentName, path); break; case SET: - mManager.setTagsOrAttributesValue(alterMap, path.getFullPath()); + mManager.setTagsOrAttributesValue(alterMap, path); break; case DROP: - mManager.dropTagsOrAttributes(alterMap.keySet(), path.getFullPath()); + mManager.dropTagsOrAttributes(alterMap.keySet(), path); break; case ADD_TAGS: - mManager.addTags(alterMap, path.getFullPath()); + mManager.addTags(alterMap, path); break; case ADD_ATTRIBUTES: - mManager.addAttributes(alterMap, path.getFullPath()); + mManager.addAttributes(alterMap, path); break; case UPSERT: mManager.upsertTagsAndAttributes(alterTimeSeriesPlan.getAlias(), alterTimeSeriesPlan.getTagsMap(), alterTimeSeriesPlan.getAttributesMap(), - path.getFullPath()); + path); break; } } catch (MetadataException e) { @@ -1029,16 +1031,16 @@ public class PlanExecutor implements IPlanExecutor { } catch (IOException e) { throw new QueryProcessException(String .format("Something went wrong while read/write the [%s]'s tag/attribute info.", - path.getFullPath())); + path.toString())); } return true; } public boolean setStorageGroup(SetStorageGroupPlan setStorageGroupPlan) throws QueryProcessException { - Path path = setStorageGroupPlan.getPath(); + PartialPath path = setStorageGroupPlan.getPath(); try { - mManager.setStorageGroup(path.getFullPath()); + mManager.setStorageGroup(path); } catch (MetadataException e) { throw new QueryProcessException(e); } @@ -1047,11 +1049,11 @@ public class PlanExecutor implements IPlanExecutor { protected boolean deleteStorageGroups(DeleteStorageGroupPlan deleteStorageGroupPlan) throws QueryProcessException { - List<String> deletePathList = new ArrayList<>(); + List<PartialPath> deletePathList = new ArrayList<>(); try { - for (Path storageGroupPath : deleteStorageGroupPlan.getPaths()) { - StorageEngine.getInstance().deleteStorageGroup(storageGroupPath.getFullPath()); - deletePathList.add(storageGroupPath.getFullPath()); + for (PartialPath storageGroupPath : deleteStorageGroupPlan.getPaths()) { + StorageEngine.getInstance().deleteStorageGroup(storageGroupPath); + deletePathList.add(storageGroupPath); } mManager.deleteStorageGroups(deletePathList); } catch (MetadataException e) { @@ -1065,9 +1067,9 @@ public class PlanExecutor implements IPlanExecutor { * * @param pathList deleted paths */ - protected void deleteDataOfTimeSeries(List<Path> pathList) + protected void deleteDataOfTimeSeries(List<PartialPath> pathList) throws QueryProcessException, StorageGroupNotSetException, StorageEngineException { - for (Path p : pathList) { + for (PartialPath p : pathList) { DeletePlan deletePlan = new DeletePlan(); deletePlan.addPath(p); deletePlan.setDeleteStartTime(Long.MIN_VALUE); @@ -1081,7 +1083,7 @@ public class PlanExecutor implements IPlanExecutor { AuthorType authorType = plan.getAuthorType(); String userName = plan.getUserName(); String roleName = plan.getRoleName(); - Path path = plan.getNodeName(); + PartialPath path = plan.getNodeName(); ListDataSet dataSet; @@ -1197,7 +1199,7 @@ public class PlanExecutor implements IPlanExecutor { } } - private ListDataSet executeListRolePrivileges(String roleName, Path path) throws AuthException { + private ListDataSet executeListRolePrivileges(String roleName, PartialPath path) throws AuthException { Role role = authorizer.getRole(roleName); if (role != null) { List<Path> headerList = new ArrayList<>(); @@ -1207,7 +1209,7 @@ public class PlanExecutor implements IPlanExecutor { ListDataSet dataSet = new ListDataSet(headerList, typeList); int index = 0; for (PathPrivilege pathPrivilege : role.getPrivilegeList()) { - if (path == null || AuthUtils.pathBelongsTo(path.getFullPath(), pathPrivilege.getPath())) { + if (path == null || AuthUtils.pathBelongsTo(path.toString(), pathPrivilege.getPath())) { RowRecord record = new RowRecord(index++); Field field = new Field(TSDataType.TEXT); field.setBinaryV(new Binary(pathPrivilege.toString())); @@ -1221,7 +1223,7 @@ public class PlanExecutor implements IPlanExecutor { } } - private ListDataSet executeListUserPrivileges(String userName, Path path) throws AuthException { + private ListDataSet executeListUserPrivileges(String userName, PartialPath path) throws AuthException { User user = authorizer.getUser(userName); if (user == null) { throw new AuthException("No such user : " + userName); @@ -1235,7 +1237,7 @@ public class PlanExecutor implements IPlanExecutor { ListDataSet dataSet = new ListDataSet(headerList, typeList); int index = 0; for (PathPrivilege pathPrivilege : user.getPrivilegeList()) { - if (path == null || AuthUtils.pathBelongsTo(path.getFullPath(), pathPrivilege.getPath())) { + if (path == null || AuthUtils.pathBelongsTo(path.toString(), pathPrivilege.getPath())) { RowRecord record = new RowRecord(index++); Field roleF = new Field(TSDataType.TEXT); roleF.setBinaryV(new Binary("")); @@ -1252,7 +1254,7 @@ public class PlanExecutor implements IPlanExecutor { continue; } for (PathPrivilege pathPrivilege : role.getPrivilegeList()) { - if (path == null || AuthUtils.pathBelongsTo(path.getFullPath(), pathPrivilege.getPath())) { + if (path == null || AuthUtils.pathBelongsTo(path.toString(), pathPrivilege.getPath())) { RowRecord record = new RowRecord(index++); Field roleF = new Field(TSDataType.TEXT); roleF.setBinaryV(new Binary(roleN)); @@ -1267,7 +1269,7 @@ public class PlanExecutor implements IPlanExecutor { return dataSet; } - protected String deleteTimeSeries(String path) throws MetadataException { + protected String deleteTimeSeries(PartialPath path) throws MetadataException { return mManager.deleteTimeseries(path); } diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/BasicFunctionOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/BasicFunctionOperator.java index b0d0521..0d6d0c5 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/BasicFunctionOperator.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/BasicFunctionOperator.java @@ -27,6 +27,7 @@ import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.qp.constant.SQLConstant; import org.apache.iotdb.db.qp.logical.Operator; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.read.common.Path; import org.apache.iotdb.tsfile.read.expression.IUnaryExpression; import org.apache.iotdb.tsfile.utils.Binary; import org.apache.iotdb.tsfile.utils.Pair; @@ -86,22 +87,22 @@ public class BasicFunctionOperator extends FunctionOperator { switch (type) { case INT32: - ret = funcToken.getUnaryExpression(singlePath, Integer.valueOf(value)); + ret = funcToken.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), Integer.valueOf(value)); break; case INT64: - ret = funcToken.getUnaryExpression(singlePath, Long.valueOf(value)); + ret = funcToken.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), Long.valueOf(value)); break; case BOOLEAN: - ret = funcToken.getUnaryExpression(singlePath, Boolean.valueOf(value)); + ret = funcToken.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), Boolean.valueOf(value)); break; case FLOAT: - ret = funcToken.getUnaryExpression(singlePath, Float.valueOf(value)); + ret = funcToken.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), Float.valueOf(value)); break; case DOUBLE: - ret = funcToken.getUnaryExpression(singlePath, Double.valueOf(value)); + ret = funcToken.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), Double.valueOf(value)); break; case TEXT: - ret = funcToken.getUnaryExpression(singlePath, + ret = funcToken.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), (value.startsWith("'") && value.endsWith("'")) || (value.startsWith("\"") && value .endsWith("\"")) ? new Binary(value.substring(1, value.length() - 1)) : new Binary(value)); diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/BasicOperatorType.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/BasicOperatorType.java index d7f7d2c..ccf6a14 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/BasicOperatorType.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/BasicOperatorType.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.qp.logical.crud; import org.apache.iotdb.db.exception.query.LogicalOperatorException; import org.apache.iotdb.db.exception.runtime.SQLParserException; +import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.qp.constant.SQLConstant; import org.apache.iotdb.tsfile.read.common.Path; import org.apache.iotdb.tsfile.read.expression.IUnaryExpression; diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/InOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/InOperator.java index a1a6e00..5365f5a 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/InOperator.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/InOperator.java @@ -93,35 +93,35 @@ public class InOperator extends FunctionOperator { for (String val : values) { integerValues.add(Integer.valueOf(val)); } - ret = In.getUnaryExpression(singlePath, integerValues, not); + ret = In.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), integerValues, not); break; case INT64: Set<Long> longValues = new HashSet<>(); for (String val : values) { longValues.add(Long.valueOf(val)); } - ret = In.getUnaryExpression(singlePath, longValues, not); + ret = In.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), longValues, not); break; case BOOLEAN: Set<Boolean> booleanValues = new HashSet<>(); for (String val : values) { booleanValues.add(Boolean.valueOf(val)); } - ret = In.getUnaryExpression(singlePath, booleanValues, not); + ret = In.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), booleanValues, not); break; case FLOAT: Set<Float> floatValues = new HashSet<>(); for (String val : values) { floatValues.add(Float.parseFloat(val)); } - ret = In.getUnaryExpression(singlePath, floatValues, not); + ret = In.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), floatValues, not); break; case DOUBLE: Set<Double> doubleValues = new HashSet<>(); for (String val : values) { doubleValues.add(Double.parseDouble(val)); } - ret = In.getUnaryExpression(singlePath, doubleValues, not); + ret = In.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), doubleValues, not); break; case TEXT: Set<Binary> binaryValues = new HashSet<>(); @@ -131,7 +131,7 @@ public class InOperator extends FunctionOperator { .endsWith("\"")) ? new Binary(val.substring(1, val.length() - 1)) : new Binary(val)); } - ret = In.getUnaryExpression(singlePath, binaryValues, not); + ret = In.getUnaryExpression(new Path(singlePath.getPathWithoutLastNode(), singlePath.getLastNode()), binaryValues, not); break; default: throw new LogicalOperatorException(type.toString(), ""); diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePartitionPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePartitionPlan.java index c720d27..4ddc8b1 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePartitionPlan.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePartitionPlan.java @@ -21,27 +21,27 @@ package org.apache.iotdb.db.qp.physical.crud; import java.util.List; import java.util.Set; +import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.qp.logical.Operator.OperatorType; import org.apache.iotdb.db.qp.physical.PhysicalPlan; -import org.apache.iotdb.tsfile.read.common.Path; public class DeletePartitionPlan extends PhysicalPlan { - private String storageGroupName; + private PartialPath storageGroupName; private Set<Long> partitionId; - public DeletePartitionPlan(String storageGroupName, Set<Long> partitionId) { + public DeletePartitionPlan(PartialPath storageGroupName, Set<Long> partitionId) { super(false, OperatorType.DELETE_PARTITION); this.storageGroupName = storageGroupName; this.partitionId = partitionId; } @Override - public List<Path> getPaths() { + public List<PartialPath> getPaths() { return null; } - public String getStorageGroupName() { + public PartialPath getStorageGroupName() { return storageGroupName; } diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java index 2f65c75..d916861 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java +++ b/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java @@ -36,6 +36,7 @@ import org.apache.iotdb.db.engine.StorageEngine; import org.apache.iotdb.db.engine.querycontext.QueryDataSource; import org.apache.iotdb.db.engine.storagegroup.TsFileResource; import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.query.context.QueryContext; import org.apache.iotdb.db.query.externalsort.serialize.IExternalSortFileDeserializer; @@ -114,7 +115,8 @@ public class QueryResourceManager { } public QueryDataSource getQueryDataSource(Path selectedPath, - QueryContext context, Filter filter) throws StorageEngineException, QueryProcessException { + QueryContext context, Filter filter) + throws StorageEngineException, QueryProcessException, IllegalPathException { SingleSeriesExpression singleSeriesExpression = new SingleSeriesExpression(selectedPath, filter); diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java index 6393f3b..7711292 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java +++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.query.dataset.groupby; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan; import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan; @@ -68,7 +69,7 @@ public class GroupByWithValueFilterDataSet extends GroupByEngineDataSet { * constructor. */ public GroupByWithValueFilterDataSet(QueryContext context, GroupByTimePlan groupByTimePlan) - throws StorageEngineException, QueryProcessException { + throws StorageEngineException, QueryProcessException, IllegalPathException { super(context, groupByTimePlan); this.timeStampFetchSize = IoTDBDescriptor.getInstance().getConfig().getBatchSize(); initGroupBy(context, groupByTimePlan); @@ -84,7 +85,7 @@ public class GroupByWithValueFilterDataSet extends GroupByEngineDataSet { * init reader and aggregate function. */ protected void initGroupBy(QueryContext context, GroupByTimePlan groupByTimePlan) - throws StorageEngineException, QueryProcessException { + throws StorageEngineException, QueryProcessException, IllegalPathException { this.timestampGenerator = getTimeGenerator(groupByTimePlan.getExpression(), context, groupByTimePlan); this.allDataReaderList = new ArrayList<>(); this.groupByTimePlan = groupByTimePlan; @@ -101,7 +102,7 @@ public class GroupByWithValueFilterDataSet extends GroupByEngineDataSet { protected IReaderByTimestamp getReaderByTime(Path path, RawDataQueryPlan queryPlan, TSDataType dataType, QueryContext context, TsFileFilter fileFilter) - throws StorageEngineException, QueryProcessException { + throws StorageEngineException, QueryProcessException, IllegalPathException { return new SeriesReaderByTimestamp(path, queryPlan.getAllMeasurementsInDevice(path.getDevice()), dataType, context, QueryResourceManager.getInstance().getQueryDataSource(path, context, null), fileFilter); } diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java index 25db91a..d0b3198 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java +++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.query.dataset.groupby; import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan; import org.apache.iotdb.db.query.aggregation.AggregateResult; @@ -66,14 +67,14 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet { * constructor. */ public GroupByWithoutValueFilterDataSet(QueryContext context, GroupByTimePlan groupByTimePlan) - throws StorageEngineException, QueryProcessException { + throws StorageEngineException, QueryProcessException, IllegalPathException { super(context, groupByTimePlan); initGroupBy(context, groupByTimePlan); } protected void initGroupBy(QueryContext context, GroupByTimePlan groupByTimePlan) - throws StorageEngineException, QueryProcessException { + throws StorageEngineException, QueryProcessException, IllegalPathException { IExpression expression = groupByTimePlan.getExpression(); Filter timeFilter = null; @@ -139,7 +140,7 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet { protected GroupByExecutor getGroupByExecutor(Path path, Set<String> allSensors, TSDataType dataType, QueryContext context, Filter timeFilter, TsFileFilter fileFilter) - throws StorageEngineException, QueryProcessException { + throws StorageEngineException, QueryProcessException, IllegalPathException { return new LocalGroupByExecutor(path, allSensors, dataType, context, timeFilter, fileFilter); } } \ No newline at end of file diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java index 3ccd596..3b60e27 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java +++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.query.dataset.groupby; import org.apache.iotdb.db.engine.querycontext.QueryDataSource; import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.query.aggregation.AggregateResult; import org.apache.iotdb.db.query.context.QueryContext; @@ -57,7 +58,7 @@ public class LocalGroupByExecutor implements GroupByExecutor { public LocalGroupByExecutor(Path path, Set<String> allSensors, TSDataType dataType, QueryContext context, Filter timeFilter, TsFileFilter fileFilter) - throws StorageEngineException, QueryProcessException { + throws StorageEngineException, QueryProcessException, IllegalPathException { queryDataSource = QueryResourceManager.getInstance() .getQueryDataSource(path, context, timeFilter); // update filter by TTL diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java b/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java index d04a24f..b39bb6a 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java +++ b/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.query.executor; import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.qp.physical.crud.*; import org.apache.iotdb.db.query.context.QueryContext; @@ -47,7 +48,7 @@ public interface IQueryRouter { */ QueryDataSet groupBy(GroupByTimePlan groupByTimePlan, QueryContext context) throws QueryFilterOptimizationException, StorageEngineException, - QueryProcessException, IOException; + QueryProcessException, IOException, IllegalPathException; /** * Execute fill query. @@ -60,7 +61,7 @@ public interface IQueryRouter { */ QueryDataSet groupByFill(GroupByTimeFillPlan groupByFillPlan, QueryContext context) throws QueryFilterOptimizationException, StorageEngineException, - QueryProcessException, IOException; + QueryProcessException, IOException, IllegalPathException; /** * Execute last query diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java b/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java index c2591d2..c14f62a 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java +++ b/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java @@ -19,12 +19,26 @@ package org.apache.iotdb.db.query.executor; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.exception.query.QueryProcessException; -import org.apache.iotdb.db.qp.physical.crud.*; +import org.apache.iotdb.db.metadata.PartialPath; +import org.apache.iotdb.db.qp.physical.crud.AggregationPlan; +import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan; +import org.apache.iotdb.db.qp.physical.crud.GroupByTimeFillPlan; +import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan; +import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan; +import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan; import org.apache.iotdb.db.query.context.QueryContext; -import org.apache.iotdb.db.query.dataset.SingleDataSet; -import org.apache.iotdb.db.query.dataset.groupby.*; +import org.apache.iotdb.db.query.dataset.groupby.GroupByEngineDataSet; +import org.apache.iotdb.db.query.dataset.groupby.GroupByFillDataSet; +import org.apache.iotdb.db.query.dataset.groupby.GroupByTimeDataSet; +import org.apache.iotdb.db.query.dataset.groupby.GroupByWithValueFilterDataSet; +import org.apache.iotdb.db.query.dataset.groupby.GroupByWithoutValueFilterDataSet; import org.apache.iotdb.db.query.executor.fill.IFill; import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; @@ -39,10 +53,6 @@ import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.util.List; -import java.util.Map; - /** * Query entrance class of IoTDB query process. All query clause will be transformed to physical * plan, physical plan will be executed by EngineQueryRouter. @@ -55,7 +65,8 @@ public class QueryRouter implements IQueryRouter { public QueryDataSet rawDataQuery(RawDataQueryPlan queryPlan, QueryContext context) throws StorageEngineException, QueryProcessException { IExpression expression = queryPlan.getExpression(); - List<Path> deduplicatedPaths = queryPlan.getDeduplicatedPaths(); + List<Path> deduplicatedPaths = queryPlan.getDeduplicatedPaths().stream().map(PartialPath::toTSFilePath).collect( + Collectors.toList()); IExpression optimizedExpression; try { @@ -97,7 +108,8 @@ public class QueryRouter implements IQueryRouter { } IExpression expression = aggregationPlan.getExpression(); - List<Path> deduplicatedPaths = aggregationPlan.getDeduplicatedPaths(); + List<Path> deduplicatedPaths = aggregationPlan.getDeduplicatedPaths().stream().map(PartialPath::toTSFilePath).collect( + Collectors.toList()); // optimize expression to an executable one IExpression optimizedExpression = @@ -126,7 +138,7 @@ public class QueryRouter implements IQueryRouter { @Override public QueryDataSet groupBy(GroupByTimePlan groupByTimePlan, QueryContext context) - throws QueryFilterOptimizationException, StorageEngineException, QueryProcessException, IOException { + throws QueryFilterOptimizationException, StorageEngineException, QueryProcessException, IOException, IllegalPathException { if (logger.isDebugEnabled()) { logger.debug("paths:" + groupByTimePlan.getPaths() + " level:" + groupByTimePlan.getLevel()); @@ -139,7 +151,8 @@ public class QueryRouter implements IQueryRouter { long endTime = groupByTimePlan.getEndTime(); IExpression expression = groupByTimePlan.getExpression(); - List<Path> selectedSeries = groupByTimePlan.getDeduplicatedPaths(); + List<Path> selectedSeries = groupByTimePlan.getDeduplicatedPaths().stream().map(PartialPath::toTSFilePath).collect( + Collectors.toList()); GlobalTimeExpression timeExpression = new GlobalTimeExpression( new GroupByFilter(unit, slidingStep, startTime, endTime)); @@ -171,12 +184,12 @@ public class QueryRouter implements IQueryRouter { } protected GroupByWithoutValueFilterDataSet getGroupByWithoutValueFilterDataSet(QueryContext context, GroupByTimePlan plan) - throws StorageEngineException, QueryProcessException { + throws StorageEngineException, QueryProcessException, IllegalPathException { return new GroupByWithoutValueFilterDataSet(context, plan); } protected GroupByWithValueFilterDataSet getGroupByWithValueFilterDataSet(QueryContext context, GroupByTimePlan plan) - throws StorageEngineException, QueryProcessException { + throws StorageEngineException, QueryProcessException, IllegalPathException { return new GroupByWithValueFilterDataSet(context, plan); } @@ -189,7 +202,8 @@ public class QueryRouter implements IQueryRouter { @Override public QueryDataSet fill(FillQueryPlan fillQueryPlan, QueryContext context) throws StorageEngineException, QueryProcessException, IOException { - List<Path> fillPaths = fillQueryPlan.getDeduplicatedPaths(); + List<Path> fillPaths = fillQueryPlan.getDeduplicatedPaths().stream().map(PartialPath::toTSFilePath).collect( + Collectors.toList()); List<TSDataType> dataTypes = fillQueryPlan.getDeduplicatedDataTypes(); long queryTime = fillQueryPlan.getQueryTime(); Map<TSDataType, IFill> fillType = fillQueryPlan.getFillType(); @@ -208,9 +222,10 @@ public class QueryRouter implements IQueryRouter { @Override public QueryDataSet groupByFill(GroupByTimeFillPlan groupByFillPlan, QueryContext context) - throws QueryFilterOptimizationException, StorageEngineException, QueryProcessException, IOException { + throws QueryFilterOptimizationException, StorageEngineException, QueryProcessException, IOException, IllegalPathException { GroupByEngineDataSet groupByEngineDataSet = (GroupByEngineDataSet) groupBy(groupByFillPlan, context); - return new GroupByFillDataSet(groupByFillPlan.getDeduplicatedPaths(), groupByFillPlan.getDeduplicatedDataTypes(), + return new GroupByFillDataSet(groupByFillPlan.getDeduplicatedPaths().stream().map(PartialPath::toTSFilePath).collect( + Collectors.toList()), groupByFillPlan.getDeduplicatedDataTypes(), groupByEngineDataSet, groupByFillPlan.getFillType(), context, groupByFillPlan); } diff --git a/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java b/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java index c6a07be..29d3922 100644 --- a/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java +++ b/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java @@ -29,6 +29,7 @@ import org.apache.iotdb.db.engine.storagegroup.TsFileResource; import org.apache.iotdb.db.exception.LoadFileException; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.exception.SyncDeviceOwnerConflictException; +import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.sync.conf.SyncConstant; import org.apache.iotdb.db.utils.FileLoaderUtils; import org.slf4j.Logger; @@ -139,7 +140,7 @@ public class FileLoader implements IFileLoader { StorageEngine.getInstance().loadNewTsFileForSync(tsFileResource); } catch (SyncDeviceOwnerConflictException e) { LOGGER.error("Device owner has conflicts, so skip the loading file", e); - } catch (LoadFileException | StorageEngineException e) { + } catch (LoadFileException | StorageEngineException | IllegalPathException e) { LOGGER.error("Can not load new tsfile {}", newTsFile.getAbsolutePath(), e); throw new IOException(e); }
