| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| package org.apache.iotdb.cluster.metadata; |
| |
| import org.apache.iotdb.cluster.client.async.AsyncDataClient; |
| import org.apache.iotdb.cluster.client.sync.SyncClientAdaptor; |
| import org.apache.iotdb.cluster.client.sync.SyncDataClient; |
| import org.apache.iotdb.cluster.config.ClusterDescriptor; |
| import org.apache.iotdb.cluster.coordinator.Coordinator; |
| import org.apache.iotdb.cluster.exception.CheckConsistencyException; |
| import org.apache.iotdb.cluster.exception.UnsupportedPlanException; |
| import org.apache.iotdb.cluster.partition.PartitionGroup; |
| import org.apache.iotdb.cluster.query.manage.QueryCoordinator; |
| import org.apache.iotdb.cluster.rpc.thrift.GetAllPathsResult; |
| import org.apache.iotdb.cluster.rpc.thrift.Node; |
| import org.apache.iotdb.cluster.rpc.thrift.PullSchemaRequest; |
| import org.apache.iotdb.cluster.rpc.thrift.PullSchemaResp; |
| import org.apache.iotdb.cluster.server.RaftServer; |
| import org.apache.iotdb.cluster.server.member.DataGroupMember; |
| import org.apache.iotdb.cluster.server.member.MetaGroupMember; |
| import org.apache.iotdb.cluster.utils.ClusterUtils; |
| import org.apache.iotdb.db.conf.IoTDBConstant; |
| import org.apache.iotdb.db.conf.IoTDBDescriptor; |
| import org.apache.iotdb.db.exception.metadata.IllegalPathException; |
| import org.apache.iotdb.db.exception.metadata.MetadataException; |
| import org.apache.iotdb.db.exception.metadata.PathNotExistException; |
| import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException; |
| import org.apache.iotdb.db.metadata.MManager; |
| import org.apache.iotdb.db.metadata.MetaUtils; |
| 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.constant.SQLConstant; |
| import org.apache.iotdb.db.qp.physical.PhysicalPlan; |
| import org.apache.iotdb.db.qp.physical.crud.InsertMultiTabletPlan; |
| import org.apache.iotdb.db.qp.physical.crud.InsertPlan; |
| import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan; |
| import org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan; |
| import org.apache.iotdb.db.qp.physical.crud.InsertRowsPlan; |
| import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan; |
| import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan; |
| import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan; |
| import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan; |
| import org.apache.iotdb.db.qp.physical.sys.ShowDevicesPlan; |
| import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan; |
| import org.apache.iotdb.db.query.context.QueryContext; |
| import org.apache.iotdb.db.query.dataset.ShowDevicesResult; |
| import org.apache.iotdb.db.query.dataset.ShowTimeSeriesResult; |
| import org.apache.iotdb.db.service.IoTDB; |
| import org.apache.iotdb.db.utils.SchemaUtils; |
| import org.apache.iotdb.db.utils.TypeInferenceUtils; |
| import org.apache.iotdb.rpc.TSStatusCode; |
| import org.apache.iotdb.service.rpc.thrift.TSStatus; |
| import org.apache.iotdb.tsfile.common.cache.LRUCache; |
| import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; |
| import org.apache.iotdb.tsfile.common.constant.TsFileConstant; |
| import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; |
| import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; |
| import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; |
| import org.apache.iotdb.tsfile.read.TimeValuePair; |
| import org.apache.iotdb.tsfile.utils.Pair; |
| import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; |
| import org.apache.iotdb.tsfile.write.schema.TimeseriesSchema; |
| |
| import org.apache.thrift.TException; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.ByteArrayOutputStream; |
| import java.io.DataOutputStream; |
| import java.io.IOException; |
| import java.lang.reflect.Array; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| import java.util.concurrent.ConcurrentSkipListSet; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.LinkedBlockingDeque; |
| import java.util.concurrent.ThreadPoolExecutor; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| import java.util.function.Supplier; |
| import java.util.stream.Collectors; |
| |
| import static org.apache.iotdb.cluster.query.ClusterPlanExecutor.LOG_FAIL_CONNECT; |
| import static org.apache.iotdb.cluster.query.ClusterPlanExecutor.THREAD_POOL_SIZE; |
| import static org.apache.iotdb.cluster.query.ClusterPlanExecutor.waitForThreadPool; |
| import static org.apache.iotdb.db.utils.EncodingInferenceUtils.getDefaultEncoding; |
| |
| @SuppressWarnings("java:S1135") // ignore todos |
| public class CMManager extends MManager { |
| |
| private static final Logger logger = LoggerFactory.getLogger(CMManager.class); |
| |
| private ReentrantReadWriteLock cacheLock = new ReentrantReadWriteLock(); |
| // only cache the series who is writing, we need not to cache series who is reading |
| // because the read is slow, so pull from remote is little cost comparing to the disk io |
| private RemoteMetaCache mRemoteMetaCache; |
| private MetaPuller metaPuller; |
| private MetaGroupMember metaGroupMember; |
| private Coordinator coordinator; |
| |
| private CMManager() { |
| super(); |
| metaPuller = MetaPuller.getInstance(); |
| int remoteCacheSize = config.getmRemoteSchemaCacheSize(); |
| mRemoteMetaCache = new RemoteMetaCache(remoteCacheSize); |
| } |
| |
| private static class MManagerHolder { |
| |
| private MManagerHolder() { |
| // allowed to do nothing |
| } |
| |
| private static final CMManager INSTANCE = new CMManager(); |
| } |
| |
| /** |
| * we should not use this function in other place, but only in IoTDB class |
| * |
| * @return |
| */ |
| public static CMManager getInstance() { |
| return CMManager.MManagerHolder.INSTANCE; |
| } |
| |
| /** |
| * sync meta leader to get the newest partition table and storage groups. |
| * |
| * @throws MetadataException throws MetadataException if necessary |
| */ |
| public void syncMetaLeader() throws MetadataException { |
| try { |
| metaGroupMember.syncLeaderWithConsistencyCheck(false); |
| } catch (CheckConsistencyException e) { |
| throw new MetadataException(e); |
| } |
| } |
| |
| @Override |
| public String deleteTimeseries(PartialPath prefixPath) throws MetadataException { |
| cacheLock.writeLock().lock(); |
| mRemoteMetaCache.removeItem(prefixPath); |
| cacheLock.writeLock().unlock(); |
| return super.deleteTimeseries(prefixPath); |
| } |
| |
| @Override |
| public void deleteStorageGroups(List<PartialPath> storageGroups) throws MetadataException { |
| cacheLock.writeLock().lock(); |
| for (PartialPath storageGroup : storageGroups) { |
| mRemoteMetaCache.removeItem(storageGroup); |
| } |
| cacheLock.writeLock().unlock(); |
| super.deleteStorageGroups(storageGroups); |
| } |
| |
| @Override |
| public TSDataType getSeriesType(PartialPath path) throws MetadataException { |
| // try remote cache first |
| try { |
| cacheLock.readLock().lock(); |
| MeasurementMNode measurementMNode = mRemoteMetaCache.get(path); |
| if (measurementMNode != null) { |
| return measurementMNode.getSchema().getType(); |
| } |
| } finally { |
| cacheLock.readLock().unlock(); |
| } |
| |
| // try local MTree |
| TSDataType seriesType; |
| try { |
| seriesType = super.getSeriesType(path); |
| } catch (PathNotExistException e) { |
| // pull from remote node |
| List<MeasurementSchema> schemas = |
| metaPuller.pullMeasurementSchemas(Collections.singletonList(path)); |
| if (!schemas.isEmpty()) { |
| MeasurementSchema measurementSchema = schemas.get(0); |
| MeasurementMNode measurementMNode = |
| new MeasurementMNode( |
| null, measurementSchema.getMeasurementId(), measurementSchema, null); |
| cacheMeta(path, measurementMNode); |
| return schemas.get(0).getType(); |
| } else { |
| throw e; |
| } |
| } |
| return seriesType; |
| } |
| |
| /** |
| * the {@link org.apache.iotdb.db.writelog.recover.LogReplayer#replayLogs(Supplier)} will call |
| * this to get schema after restart we should retry to get schema util we get the schema. |
| * |
| * @param deviceId the device id. |
| * @param measurements the measurements. |
| */ |
| @Override |
| public MeasurementMNode[] getMNodes(PartialPath deviceId, String[] measurements) |
| throws MetadataException { |
| try { |
| return super.getMNodes(deviceId, measurements); |
| } catch (MetadataException e) { |
| // some measurements not exist in local |
| // try cache |
| MeasurementMNode[] measurementMNodes = new MeasurementMNode[measurements.length]; |
| int failedMeasurementIndex = getMNodesLocally(deviceId, measurements, measurementMNodes); |
| if (failedMeasurementIndex == -1) { |
| return measurementMNodes; |
| } |
| |
| // will retry util get schema |
| pullSeriesSchemas(deviceId, measurements); |
| |
| // try again |
| failedMeasurementIndex = getMNodesLocally(deviceId, measurements, measurementMNodes); |
| if (failedMeasurementIndex != -1) { |
| throw new MetadataException( |
| deviceId.getFullPath() |
| + IoTDBConstant.PATH_SEPARATOR |
| + measurements[failedMeasurementIndex] |
| + " is not found"); |
| } |
| return measurementMNodes; |
| } |
| } |
| |
| /** @return -1 if all schemas are found, or the first index of the non-exist schema */ |
| private int getMNodesLocally( |
| PartialPath deviceId, String[] measurements, MeasurementMNode[] measurementMNodes) { |
| int failedMeasurementIndex = -1; |
| cacheLock.readLock().lock(); |
| try { |
| for (int i = 0; i < measurements.length && failedMeasurementIndex == -1; i++) { |
| MeasurementMNode measurementMNode = |
| mRemoteMetaCache.get(deviceId.concatNode(measurements[i])); |
| if (measurementMNode == null) { |
| failedMeasurementIndex = i; |
| } else { |
| measurementMNodes[i] = measurementMNode; |
| } |
| } |
| } finally { |
| cacheLock.readLock().unlock(); |
| } |
| return failedMeasurementIndex; |
| } |
| |
| private void pullSeriesSchemas(PartialPath deviceId, String[] measurementList) |
| throws MetadataException { |
| List<PartialPath> schemasToPull = new ArrayList<>(); |
| for (String s : measurementList) { |
| schemasToPull.add(deviceId.concatNode(s)); |
| } |
| List<MeasurementSchema> schemas = metaPuller.pullMeasurementSchemas(schemasToPull); |
| for (MeasurementSchema schema : schemas) { |
| // TODO-Cluster: also pull alias? |
| // take care, the pulled schema's measurement Id is only series name |
| MeasurementMNode measurementMNode = |
| new MeasurementMNode(null, schema.getMeasurementId(), schema, null); |
| cacheMeta(deviceId.concatNode(schema.getMeasurementId()), measurementMNode); |
| } |
| logger.debug("Pulled {}/{} schemas from remote", schemas.size(), measurementList.length); |
| } |
| |
| @Override |
| public void cacheMeta(PartialPath seriesPath, MeasurementMNode measurementMNode) { |
| cacheLock.writeLock().lock(); |
| mRemoteMetaCache.put(seriesPath, measurementMNode); |
| cacheLock.writeLock().unlock(); |
| } |
| |
| @Override |
| public void updateLastCache( |
| PartialPath seriesPath, |
| TimeValuePair timeValuePair, |
| boolean highPriorityUpdate, |
| Long latestFlushedTime, |
| MeasurementMNode node) { |
| cacheLock.writeLock().lock(); |
| try { |
| MeasurementMNode measurementMNode = mRemoteMetaCache.get(seriesPath); |
| if (measurementMNode != null) { |
| measurementMNode.updateCachedLast(timeValuePair, highPriorityUpdate, latestFlushedTime); |
| } |
| } finally { |
| cacheLock.writeLock().unlock(); |
| } |
| // maybe local also has the timeseries |
| super.updateLastCache(seriesPath, timeValuePair, highPriorityUpdate, latestFlushedTime, node); |
| } |
| |
| @Override |
| public TimeValuePair getLastCache(PartialPath seriesPath) { |
| MeasurementMNode measurementMNode = mRemoteMetaCache.get(seriesPath); |
| if (measurementMNode != null) { |
| return measurementMNode.getCachedLast(); |
| } |
| |
| return super.getLastCache(seriesPath); |
| } |
| |
| @Override |
| public MNode getSeriesSchemasAndReadLockDevice(InsertPlan plan) throws MetadataException { |
| MeasurementMNode[] measurementMNodes = new MeasurementMNode[plan.getMeasurements().length]; |
| int nonExistSchemaIndex = |
| getMNodesLocally(plan.getDeviceId(), plan.getMeasurements(), measurementMNodes); |
| if (nonExistSchemaIndex == -1) { |
| plan.setMeasurementMNodes(measurementMNodes); |
| return new MNode(null, plan.getDeviceId().getDevice()); |
| } |
| // auto-create schema in IoTDBConfig is always disabled in the cluster version, and we have |
| // another config in ClusterConfig to do this |
| return super.getSeriesSchemasAndReadLockDevice(plan); |
| } |
| |
| @Override |
| public MeasurementSchema getSeriesSchema(PartialPath device, String measurement) |
| throws MetadataException { |
| try { |
| MeasurementSchema measurementSchema = super.getSeriesSchema(device, measurement); |
| if (measurementSchema != null) { |
| return measurementSchema; |
| } |
| } catch (PathNotExistException e) { |
| // not found in local |
| } |
| |
| // try cache |
| cacheLock.readLock().lock(); |
| try { |
| MeasurementMNode measurementMNode = mRemoteMetaCache.get(device.concatNode(measurement)); |
| if (measurementMNode != null) { |
| return measurementMNode.getSchema(); |
| } |
| } finally { |
| cacheLock.readLock().unlock(); |
| } |
| |
| // pull from remote |
| pullSeriesSchemas(device, new String[] {measurement}); |
| |
| // try again |
| cacheLock.readLock().lock(); |
| try { |
| MeasurementMNode measurementMeta = mRemoteMetaCache.get(device.concatNode(measurement)); |
| if (measurementMeta != null) { |
| return measurementMeta.getSchema(); |
| } |
| } finally { |
| cacheLock.readLock().unlock(); |
| } |
| return super.getSeriesSchema(device, measurement); |
| } |
| |
| /** |
| * Check whether the path exists. |
| * |
| * @param path a full path or a prefix path |
| */ |
| @Override |
| public boolean isPathExist(PartialPath path) { |
| boolean localExist = super.isPathExist(path); |
| if (localExist) { |
| return true; |
| } |
| |
| // search the cache |
| cacheLock.readLock().lock(); |
| try { |
| return mRemoteMetaCache.containsKey(path); |
| } finally { |
| cacheLock.readLock().unlock(); |
| } |
| } |
| |
| private static class RemoteMetaCache extends LRUCache<PartialPath, MeasurementMNode> { |
| |
| RemoteMetaCache(int cacheSize) { |
| super(cacheSize); |
| } |
| |
| @Override |
| protected MeasurementMNode loadObjectByKey(PartialPath key) { |
| return null; |
| } |
| |
| @Override |
| public synchronized void removeItem(PartialPath key) { |
| cache.keySet().removeIf(s -> s.getFullPath().startsWith(key.getFullPath())); |
| } |
| |
| @Override |
| public synchronized MeasurementMNode get(PartialPath key) { |
| try { |
| return super.get(key); |
| } catch (IOException e) { |
| // not happening |
| return null; |
| } |
| } |
| |
| public synchronized boolean containsKey(PartialPath key) { |
| return cache.containsKey(key); |
| } |
| } |
| |
| /** |
| * create storage groups for CreateTimeseriesPlan, CreateMultiTimeseriesPlan and InsertPlan, also |
| * create timeseries for InsertPlan. Only the three kind of plans can use this method. |
| */ |
| public void createSchema(PhysicalPlan plan) throws MetadataException, CheckConsistencyException { |
| List<PartialPath> storageGroups = new ArrayList<>(); |
| // for InsertPlan, try to just use deviceIds to get related storage groups because there's no |
| // need to call getPaths to concat deviceId and sensor as they will gain same result, |
| // for CreateTimeSeriesPlan, use getPath() to get timeseries to get related storage group, |
| // for CreateMultiTimeSeriesPlan, use getPaths() to get all timeseries to get related storage |
| // groups. |
| if (plan instanceof InsertRowPlan |
| || plan instanceof InsertRowsOfOneDevicePlan |
| || plan instanceof InsertTabletPlan) { |
| storageGroups.addAll( |
| getStorageGroups(Collections.singletonList(((InsertPlan) plan).getDeviceId()))); |
| } else if (plan instanceof InsertRowsPlan) { |
| storageGroups.addAll( |
| getStorageGroups( |
| ((InsertRowsPlan) plan) |
| .getInsertRowPlanList().stream() |
| .map(InsertPlan::getDeviceId) |
| .collect(Collectors.toList()))); |
| } else if (plan instanceof InsertMultiTabletPlan) { |
| storageGroups.addAll( |
| getStorageGroups( |
| ((InsertMultiTabletPlan) plan) |
| .getInsertTabletPlanList().stream() |
| .map(InsertPlan::getDeviceId) |
| .collect(Collectors.toList()))); |
| } else if (plan instanceof CreateTimeSeriesPlan) { |
| storageGroups.addAll( |
| getStorageGroups(Collections.singletonList(((CreateTimeSeriesPlan) plan).getPath()))); |
| } else { |
| storageGroups.addAll(getStorageGroups(plan.getPaths())); |
| } |
| |
| // create storage groups |
| createStorageGroups(storageGroups); |
| |
| // need to verify the storage group is created |
| verifyCreatedSgSuccess(storageGroups, plan); |
| |
| // try to create timeseries for insertPlan |
| if (plan instanceof InsertPlan && !createTimeseries((InsertPlan) plan)) { |
| throw new MetadataException("Failed to create timeseries from InsertPlan automatically."); |
| } |
| } |
| |
| /** return storage groups paths for given deviceIds or timeseries. */ |
| private List<PartialPath> getStorageGroups(List<PartialPath> paths) throws MetadataException { |
| Set<PartialPath> storageGroups = new HashSet<>(); |
| for (PartialPath path : paths) { |
| storageGroups.add( |
| MetaUtils.getStorageGroupPathByLevel( |
| path, IoTDBDescriptor.getInstance().getConfig().getDefaultStorageGroupLevel())); |
| } |
| return new ArrayList<>(storageGroups); |
| } |
| |
| @SuppressWarnings("squid:S3776") |
| private void verifyCreatedSgSuccess(List<PartialPath> storageGroups, PhysicalPlan physicalPlan) { |
| long startTime = System.currentTimeMillis(); |
| boolean[] ready = new boolean[storageGroups.size()]; |
| Arrays.fill(ready, false); |
| while (true) { |
| boolean allReady = true; |
| for (int i = 0; i < storageGroups.size(); i++) { |
| if (ready[i]) { |
| continue; |
| } |
| if (IoTDB.metaManager.isStorageGroup(storageGroups.get(i))) { |
| ready[i] = true; |
| } else { |
| allReady = false; |
| } |
| } |
| |
| if (allReady |
| || System.currentTimeMillis() - startTime |
| > ClusterDescriptor.getInstance().getConfig().getConnectionTimeoutInMS()) { |
| break; |
| } else { |
| try { |
| Thread.sleep(1); |
| } catch (InterruptedException e) { |
| logger.debug("Failed to wait for creating sgs for plan {}", physicalPlan, e); |
| Thread.currentThread().interrupt(); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Create storage groups automatically for paths. |
| * |
| * @param storageGroups the uncreated storage groups |
| */ |
| private void createStorageGroups(List<PartialPath> storageGroups) throws MetadataException { |
| for (PartialPath storageGroup : storageGroups) { |
| SetStorageGroupPlan setStorageGroupPlan = new SetStorageGroupPlan(storageGroup); |
| TSStatus setStorageGroupResult = |
| metaGroupMember.processNonPartitionedMetaPlan(setStorageGroupPlan); |
| if (setStorageGroupResult.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() |
| && setStorageGroupResult.getCode() |
| != TSStatusCode.PATH_ALREADY_EXIST_ERROR.getStatusCode()) { |
| throw new MetadataException( |
| String.format( |
| "Status Code: %d, failed to set storage group %s", |
| setStorageGroupResult.getCode(), storageGroup)); |
| } |
| } |
| } |
| |
| /** |
| * @param insertMultiTabletPlan the InsertMultiTabletPlan |
| * @return true if all InsertTabletPlan in InsertMultiTabletPlan create timeseries success, |
| * otherwise false |
| */ |
| public boolean createTimeseries(InsertMultiTabletPlan insertMultiTabletPlan) |
| throws CheckConsistencyException, IllegalPathException { |
| boolean allSuccess = true; |
| for (InsertTabletPlan insertTabletPlan : insertMultiTabletPlan.getInsertTabletPlanList()) { |
| boolean success = createTimeseries(insertTabletPlan); |
| allSuccess = allSuccess && success; |
| if (!success) { |
| logger.error( |
| "create timeseries for device={} failed, plan={}", |
| insertTabletPlan.getDeviceId(), |
| insertTabletPlan); |
| } |
| } |
| return allSuccess; |
| } |
| |
| public boolean createTimeseries(InsertRowsPlan insertRowsPlan) |
| throws CheckConsistencyException, IllegalPathException { |
| boolean allSuccess = true; |
| for (InsertRowPlan insertRowPlan : insertRowsPlan.getInsertRowPlanList()) { |
| boolean success = createTimeseries(insertRowPlan); |
| allSuccess = allSuccess && success; |
| if (!success) { |
| logger.error( |
| "create timeseries for device={} failed, plan={}", |
| insertRowPlan.getDeviceId(), |
| insertRowPlan); |
| } |
| } |
| return allSuccess; |
| } |
| |
| /** |
| * Create timeseries automatically for an InsertPlan. |
| * |
| * @param insertPlan some of the timeseries in it are not created yet |
| * @return true of all uncreated timeseries are created |
| */ |
| public boolean createTimeseries(InsertPlan insertPlan) |
| throws IllegalPathException, CheckConsistencyException { |
| if (insertPlan instanceof InsertMultiTabletPlan) { |
| return createTimeseries((InsertMultiTabletPlan) insertPlan); |
| } |
| |
| if (insertPlan instanceof InsertRowsPlan) { |
| return createTimeseries((InsertRowsPlan) insertPlan); |
| } |
| |
| List<String> seriesList = new ArrayList<>(); |
| PartialPath deviceId = insertPlan.getDeviceId(); |
| PartialPath storageGroupName; |
| try { |
| storageGroupName = |
| MetaUtils.getStorageGroupPathByLevel( |
| deviceId, IoTDBDescriptor.getInstance().getConfig().getDefaultStorageGroupLevel()); |
| } catch (MetadataException e) { |
| logger.error("Failed to infer storage group from deviceId {}", deviceId); |
| return false; |
| } |
| for (String measurementId : insertPlan.getMeasurements()) { |
| seriesList.add(deviceId.getFullPath() + TsFileConstant.PATH_SEPARATOR + measurementId); |
| } |
| PartitionGroup partitionGroup = |
| metaGroupMember.getPartitionTable().route(storageGroupName.getFullPath(), 0); |
| List<String> unregisteredSeriesList = getUnregisteredSeriesList(seriesList, partitionGroup); |
| if (unregisteredSeriesList.isEmpty()) { |
| return true; |
| } |
| logger.debug("Unregisterd series of {} are {}", seriesList, unregisteredSeriesList); |
| |
| return createTimeseries(unregisteredSeriesList, seriesList, insertPlan); |
| } |
| |
| /** |
| * create timeseries from paths in "unregisteredSeriesList". If data types are provided by the |
| * InsertPlan, use them, otherwise infer the types from the values. Use default encodings and |
| * compressions of the corresponding data type. |
| */ |
| private boolean createTimeseries( |
| List<String> unregisteredSeriesList, List<String> seriesList, InsertPlan insertPlan) |
| throws IllegalPathException { |
| List<PartialPath> paths = new ArrayList<>(); |
| List<TSDataType> dataTypes = new ArrayList<>(); |
| List<TSEncoding> encodings = new ArrayList<>(); |
| List<CompressionType> compressionTypes = new ArrayList<>(); |
| for (String seriesPath : unregisteredSeriesList) { |
| paths.add(new PartialPath(seriesPath)); |
| int index = seriesList.indexOf(seriesPath); |
| TSDataType dataType; |
| // use data types in insertPlan if provided, otherwise infer them from the values |
| if (insertPlan.getDataTypes() != null && insertPlan.getDataTypes()[index] != null) { |
| dataType = insertPlan.getDataTypes()[index]; |
| } else { |
| dataType = |
| TypeInferenceUtils.getPredictedDataType( |
| insertPlan instanceof InsertTabletPlan |
| ? Array.get(((InsertTabletPlan) insertPlan).getColumns()[index], 0) |
| : ((InsertRowPlan) insertPlan).getValues()[index], |
| true); |
| } |
| dataTypes.add(dataType); |
| // use default encoding and compression from the config |
| encodings.add(getDefaultEncoding(dataType)); |
| compressionTypes.add(TSFileDescriptor.getInstance().getConfig().getCompressor()); |
| } |
| CreateMultiTimeSeriesPlan plan = new CreateMultiTimeSeriesPlan(); |
| plan.setPaths(paths); |
| plan.setDataTypes(dataTypes); |
| plan.setEncodings(encodings); |
| plan.setCompressors(compressionTypes); |
| TSStatus result; |
| try { |
| result = coordinator.processPartitionedPlan(plan); |
| } catch (UnsupportedPlanException e) { |
| logger.error( |
| "Failed to create timeseries {} automatically. Unsupported plan exception {} ", |
| paths, |
| e.getMessage()); |
| return false; |
| } |
| if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() |
| && result.getCode() != TSStatusCode.PATH_ALREADY_EXIST_ERROR.getStatusCode() |
| && result.getCode() != TSStatusCode.NEED_REDIRECTION.getStatusCode()) { |
| logger.error( |
| "{} failed to execute create timeseries {}: {}", |
| metaGroupMember.getThisNode(), |
| paths, |
| result); |
| return false; |
| } |
| return true; |
| } |
| |
| public void setMetaGroupMember(MetaGroupMember metaGroupMember) { |
| this.metaGroupMember = metaGroupMember; |
| } |
| |
| public void setCoordinator(Coordinator coordinator) { |
| this.coordinator = coordinator; |
| } |
| |
| /** |
| * To check which timeseries in the input list is unregistered from one node in "partitionGroup". |
| */ |
| private List<String> getUnregisteredSeriesList( |
| List<String> seriesList, PartitionGroup partitionGroup) throws CheckConsistencyException { |
| if (partitionGroup.contains(metaGroupMember.getThisNode())) { |
| return getUnregisteredSeriesListLocally(seriesList, partitionGroup); |
| } else { |
| return getUnregisteredSeriesListRemotely(seriesList, partitionGroup); |
| } |
| } |
| |
| private List<String> getUnregisteredSeriesListLocally( |
| List<String> seriesList, PartitionGroup partitionGroup) throws CheckConsistencyException { |
| DataGroupMember dataMember = |
| metaGroupMember |
| .getDataClusterServer() |
| .getDataMember(partitionGroup.getHeader(), null, null); |
| return dataMember.getLocalQueryExecutor().getUnregisteredTimeseries(seriesList); |
| } |
| |
| private List<String> getUnregisteredSeriesListRemotely( |
| List<String> seriesList, PartitionGroup partitionGroup) { |
| for (Node node : partitionGroup) { |
| try { |
| List<String> result; |
| if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { |
| AsyncDataClient client = |
| metaGroupMember |
| .getClientProvider() |
| .getAsyncDataClient(node, RaftServer.getReadOperationTimeoutMS()); |
| result = |
| SyncClientAdaptor.getUnregisteredMeasurements( |
| client, partitionGroup.getHeader(), seriesList); |
| } else { |
| try (SyncDataClient syncDataClient = |
| metaGroupMember |
| .getClientProvider() |
| .getSyncDataClient(node, RaftServer.getReadOperationTimeoutMS())) { |
| result = |
| syncDataClient.getUnregisteredTimeseries(partitionGroup.getHeader(), seriesList); |
| } |
| } |
| if (result != null) { |
| return result; |
| } |
| } catch (TException | IOException e) { |
| logger.error( |
| "{}: cannot getting unregistered {} and other {} paths from {}", |
| metaGroupMember.getName(), |
| seriesList.get(0), |
| seriesList.get(seriesList.size() - 1), |
| node, |
| e); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.error( |
| "{}: getting unregistered series list {} ... {} is interrupted from {}", |
| metaGroupMember.getName(), |
| seriesList.get(0), |
| seriesList.get(seriesList.size() - 1), |
| node, |
| e); |
| } |
| } |
| return Collections.emptyList(); |
| } |
| |
| /** |
| * Pull the all timeseries schemas of given prefixPaths from remote nodes. All prefixPaths must |
| * contain a storage group. The pulled schemas will be cache in CMManager. |
| * |
| * @param ignoredGroup do not pull schema from the group to avoid backward dependency. If a user |
| * send an insert request before registering schemas, then this method may pull schemas from |
| * the same groups. If this method is called by an applier, it holds the lock of LogManager, |
| * while the pulling thread may want this lock too, resulting in a deadlock. |
| */ |
| public void pullTimeSeriesSchemas(List<PartialPath> prefixPaths, Node ignoredGroup) |
| throws MetadataException { |
| logger.debug( |
| "{}: Pulling timeseries schemas of {}, ignored group {}", |
| metaGroupMember.getName(), |
| prefixPaths, |
| ignoredGroup); |
| // split the paths by the data groups that should hold them |
| Map<PartitionGroup, List<String>> partitionGroupPathMap = new HashMap<>(); |
| for (PartialPath prefixPath : prefixPaths) { |
| if (SQLConstant.RESERVED_TIME.equalsIgnoreCase(prefixPath.getFullPath())) { |
| continue; |
| } |
| PartitionGroup partitionGroup = |
| ClusterUtils.partitionByPathTimeWithSync(prefixPath, metaGroupMember); |
| if (!partitionGroup.getHeader().equals(ignoredGroup)) { |
| partitionGroupPathMap |
| .computeIfAbsent(partitionGroup, g -> new ArrayList<>()) |
| .add(prefixPath.getFullPath()); |
| } |
| } |
| |
| // pull timeseries schema from every group involved |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: pulling schemas of {} and other {} paths from {} groups", |
| metaGroupMember.getName(), |
| prefixPaths.get(0), |
| prefixPaths.size() - 1, |
| partitionGroupPathMap.size()); |
| } |
| for (Entry<PartitionGroup, List<String>> partitionGroupListEntry : |
| partitionGroupPathMap.entrySet()) { |
| PartitionGroup partitionGroup = partitionGroupListEntry.getKey(); |
| List<String> paths = partitionGroupListEntry.getValue(); |
| pullTimeSeriesSchemas(partitionGroup, paths); |
| } |
| } |
| |
| /** |
| * Pull timeseries schemas of "prefixPaths" from "partitionGroup". If this node is a member of |
| * "partitionGroup", synchronize with the group leader and collect local schemas. Otherwise pull |
| * schemas from one node in the group. The pulled schemas will be cached in CMManager. |
| */ |
| private void pullTimeSeriesSchemas(PartitionGroup partitionGroup, List<String> prefixPaths) { |
| if (partitionGroup.contains(metaGroupMember.getThisNode())) { |
| // the node is in the target group, synchronize with leader should be enough |
| try { |
| metaGroupMember |
| .getLocalDataMember(partitionGroup.getHeader(), "Pull timeseries of " + prefixPaths) |
| .syncLeader(null); |
| } catch (CheckConsistencyException e) { |
| logger.warn("Failed to check consistency.", e); |
| } |
| return; |
| } |
| |
| // pull schemas from a remote node |
| PullSchemaRequest pullSchemaRequest = new PullSchemaRequest(); |
| pullSchemaRequest.setHeader(partitionGroup.getHeader()); |
| pullSchemaRequest.setPrefixPaths(prefixPaths); |
| |
| // decide the node access order with the help of QueryCoordinator |
| List<Node> nodes = QueryCoordinator.getINSTANCE().reorderNodes(partitionGroup); |
| for (Node node : nodes) { |
| if (tryPullTimeSeriesSchemas(node, pullSchemaRequest)) { |
| break; |
| } |
| } |
| } |
| |
| /** |
| * send the PullSchemaRequest to "node" and cache the results in CMManager if they are |
| * successfully returned. |
| * |
| * @return true if the pull succeeded, false otherwise |
| */ |
| private boolean tryPullTimeSeriesSchemas(Node node, PullSchemaRequest request) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: Pulling timeseries schemas of {} and other {} paths from {}", |
| metaGroupMember.getName(), |
| request.getPrefixPaths().get(0), |
| request.getPrefixPaths().size() - 1, |
| node); |
| } |
| |
| List<TimeseriesSchema> schemas = null; |
| try { |
| schemas = pullTimeSeriesSchemas(node, request); |
| } catch (IOException | TException e) { |
| logger.error( |
| "{}: Cannot pull timeseries schemas of {} and other {} paths from {}", |
| metaGroupMember.getName(), |
| request.getPrefixPaths().get(0), |
| request.getPrefixPaths().size() - 1, |
| node, |
| e); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.error( |
| "{}: Cannot pull timeseries schemas of {} and other {} paths from {}", |
| metaGroupMember.getName(), |
| request.getPrefixPaths().get(0), |
| request.getPrefixPaths().size() - 1, |
| node, |
| e); |
| } |
| |
| if (schemas != null) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: Pulled {} timeseries schemas of {} and other {} paths from {} of {}", |
| metaGroupMember.getName(), |
| schemas.size(), |
| request.getPrefixPaths().get(0), |
| request.getPrefixPaths().size() - 1, |
| node, |
| request.getHeader()); |
| } |
| for (TimeseriesSchema schema : schemas) { |
| SchemaUtils.cacheTimeseriesSchema(schema); |
| } |
| return true; |
| } |
| return false; |
| } |
| |
| /** |
| * send a PullSchemaRequest to a node to pull TimeseriesSchemas, and return the pulled schema or |
| * null if there was a timeout. |
| */ |
| private List<TimeseriesSchema> pullTimeSeriesSchemas(Node node, PullSchemaRequest request) |
| throws TException, InterruptedException, IOException { |
| List<TimeseriesSchema> schemas; |
| if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { |
| AsyncDataClient client = |
| metaGroupMember |
| .getClientProvider() |
| .getAsyncDataClient(node, RaftServer.getReadOperationTimeoutMS()); |
| schemas = SyncClientAdaptor.pullTimeseriesSchema(client, request); |
| } else { |
| try (SyncDataClient syncDataClient = |
| metaGroupMember |
| .getClientProvider() |
| .getSyncDataClient(node, RaftServer.getReadOperationTimeoutMS())) { |
| |
| PullSchemaResp pullSchemaResp = syncDataClient.pullTimeSeriesSchema(request); |
| ByteBuffer buffer = pullSchemaResp.schemaBytes; |
| int size = buffer.getInt(); |
| schemas = new ArrayList<>(size); |
| for (int i = 0; i < size; i++) { |
| schemas.add(TimeseriesSchema.deserializeFrom(buffer)); |
| } |
| } |
| } |
| |
| return schemas; |
| } |
| |
| /** |
| * Get the data types of "paths". If "aggregation" is not null, every path will use the |
| * aggregation. First get types locally and if some paths does not exists, pull them from other |
| * nodes. |
| * |
| * @return the left one of the pair is the column types (considering aggregation), the right one |
| * of the pair is the measurement types (not considering aggregation) |
| */ |
| public Pair<List<TSDataType>, List<TSDataType>> getSeriesTypesByPaths( |
| List<PartialPath> pathStrs, String aggregation) throws MetadataException { |
| try { |
| return getSeriesTypesByPathsLocally(pathStrs, aggregation); |
| } catch (PathNotExistException e) { |
| // pull schemas remotely and cache them |
| pullTimeSeriesSchemas(pathStrs, null); |
| return getSeriesTypesByPathsLocally(pathStrs, aggregation); |
| } |
| } |
| |
| private Pair<List<TSDataType>, List<TSDataType>> getSeriesTypesByPathsLocally( |
| List<PartialPath> pathStrs, String aggregation) throws MetadataException { |
| List<TSDataType> measurementDataTypes = |
| SchemaUtils.getSeriesTypesByPaths(pathStrs, (String) null); |
| // if the aggregation function is null, the type of column in result set |
| // is equal to the real type of the measurement |
| if (aggregation == null) { |
| return new Pair<>(measurementDataTypes, measurementDataTypes); |
| } else { |
| // if the aggregation function is not null, |
| // we should recalculate the type of column in result set |
| List<TSDataType> columnDataTypes = |
| SchemaUtils.getAggregatedDataTypes(measurementDataTypes, aggregation); |
| return new Pair<>(columnDataTypes, measurementDataTypes); |
| } |
| } |
| |
| /** |
| * Get the data types of "paths". If "aggregations" is not null, each one of it correspond to one |
| * in "paths". First get types locally and if some paths does not exists, pull them from other |
| * nodes. |
| * |
| * @param aggregations nullable, when not null, correspond to "paths" one-to-one. |
| * @return the left one of the pair is the column types (considering aggregation), the right one |
| * of the pair is the measurement types (not considering aggregation) |
| */ |
| public Pair<List<TSDataType>, List<TSDataType>> getSeriesTypesByPath( |
| List<PartialPath> paths, List<String> aggregations) throws MetadataException { |
| try { |
| return getSeriesTypesByPathLocally(paths, aggregations); |
| } catch (PathNotExistException e) { |
| return getSeriesTypesByPathRemotely(paths, aggregations); |
| } |
| } |
| |
| /** |
| * get data types of the given paths considering the aggregations from CMManger. |
| * |
| * @param aggregations nullable, when not null, correspond to "paths" one-to-one. |
| * @return the left one of the pair is the column types (considering aggregation), the right one |
| * of the pair is the measurement types (not considering aggregation) |
| */ |
| private Pair<List<TSDataType>, List<TSDataType>> getSeriesTypesByPathLocally( |
| List<PartialPath> paths, List<String> aggregations) throws MetadataException { |
| List<TSDataType> measurementDataTypes = SchemaUtils.getSeriesTypesByPaths(paths); |
| // if the aggregation function is null, the type of column in result set |
| // is equal to the real type of the measurement |
| if (aggregations == null) { |
| return new Pair<>(measurementDataTypes, measurementDataTypes); |
| } else { |
| // if the aggregation function is not null, |
| // we should recalculate the type of column in result set |
| List<TSDataType> columnDataTypes = SchemaUtils.getSeriesTypesByPaths(paths, aggregations); |
| return new Pair<>(columnDataTypes, measurementDataTypes); |
| } |
| } |
| |
| /** |
| * pull schemas from remote nodes and cache them, then get data types of the given paths |
| * considering the aggregations from CMManger. |
| * |
| * @param aggregations nullable, when not null, correspond to "paths" one-to-one. |
| * @return the left one of the pair is the column types (considering aggregation), the right one |
| * of the pair is the measurement types (not considering aggregation) |
| */ |
| private Pair<List<TSDataType>, List<TSDataType>> getSeriesTypesByPathRemotely( |
| List<PartialPath> paths, List<String> aggregations) throws MetadataException { |
| // pull schemas remotely and cache them |
| pullTimeSeriesSchemas(paths, null); |
| |
| return getSeriesTypesByPathLocally(paths, aggregations); |
| } |
| |
| /** |
| * Get all devices after removing wildcards in the path |
| * |
| * @param originPath a path potentially with wildcard |
| * @return all paths after removing wildcards in the path |
| */ |
| public Set<PartialPath> getMatchedDevices(PartialPath originPath) throws MetadataException { |
| // get all storage groups this path may belong to |
| // the key is the storage group name and the value is the path to be queried with storage group |
| // added, e.g: |
| // "root.*" will be translated into: |
| // "root.group1" -> "root.group1.*", "root.group2" -> "root.group2.*" ... |
| Map<String, String> sgPathMap = determineStorageGroup(originPath); |
| Set<PartialPath> ret = getMatchedDevices(sgPathMap); |
| logger.debug("The devices of path {} are {}", originPath, ret); |
| return ret; |
| } |
| |
| /** |
| * Split the paths by the data group they belong to and query them from the groups separately. |
| * |
| * @param sgPathMap the key is the storage group name and the value is the path to be queried with |
| * storage group added |
| * @return a collection of all queried paths |
| */ |
| private List<PartialPath> getMatchedPaths(Map<String, String> sgPathMap, boolean withAlias) |
| throws MetadataException { |
| List<PartialPath> result = new ArrayList<>(); |
| // split the paths by the data group they belong to |
| Map<PartitionGroup, List<String>> groupPathMap = new HashMap<>(); |
| for (Entry<String, String> sgPathEntry : sgPathMap.entrySet()) { |
| String storageGroupName = sgPathEntry.getKey(); |
| PartialPath pathUnderSG = new PartialPath(sgPathEntry.getValue()); |
| // find the data group that should hold the timeseries schemas of the storage group |
| PartitionGroup partitionGroup = |
| metaGroupMember.getPartitionTable().route(storageGroupName, 0); |
| if (partitionGroup.contains(metaGroupMember.getThisNode())) { |
| // this node is a member of the group, perform a local query after synchronizing with the |
| // leader |
| try { |
| metaGroupMember.getLocalDataMember(partitionGroup.getHeader()).syncLeader(null); |
| } catch (CheckConsistencyException e) { |
| logger.warn("Failed to check consistency.", e); |
| } |
| List<PartialPath> allTimeseriesName = getMatchedPathsLocally(pathUnderSG, withAlias); |
| logger.debug( |
| "{}: get matched paths of {} locally, result {}", |
| metaGroupMember.getName(), |
| partitionGroup, |
| allTimeseriesName); |
| result.addAll(allTimeseriesName); |
| } else { |
| // batch the queries of the same group to reduce communication |
| groupPathMap |
| .computeIfAbsent(partitionGroup, p -> new ArrayList<>()) |
| .add(pathUnderSG.getFullPath()); |
| } |
| } |
| |
| // query each data group separately |
| for (Entry<PartitionGroup, List<String>> partitionGroupPathEntry : groupPathMap.entrySet()) { |
| PartitionGroup partitionGroup = partitionGroupPathEntry.getKey(); |
| List<String> pathsToQuery = partitionGroupPathEntry.getValue(); |
| result.addAll(getMatchedPaths(partitionGroup, pathsToQuery, withAlias)); |
| } |
| |
| return result; |
| } |
| |
| private List<PartialPath> getMatchedPathsLocally(PartialPath partialPath, boolean withAlias) |
| throws MetadataException { |
| if (!withAlias) { |
| return getAllTimeseriesPath(partialPath); |
| } else { |
| return super.getAllTimeseriesPathWithAlias(partialPath, -1, -1).left; |
| } |
| } |
| |
| private List<PartialPath> getMatchedPaths( |
| PartitionGroup partitionGroup, List<String> pathsToQuery, boolean withAlias) |
| throws MetadataException { |
| // choose the node with lowest latency or highest throughput |
| List<Node> coordinatedNodes = QueryCoordinator.getINSTANCE().reorderNodes(partitionGroup); |
| for (Node node : coordinatedNodes) { |
| try { |
| List<PartialPath> paths = |
| getMatchedPaths(node, partitionGroup.getHeader(), pathsToQuery, withAlias); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: get matched paths of {} and other {} paths from {} in {}, result {}", |
| metaGroupMember.getName(), |
| pathsToQuery.get(0), |
| pathsToQuery.size() - 1, |
| node, |
| partitionGroup.getHeader(), |
| paths); |
| } |
| if (paths != null) { |
| // a non-null result contains correct result even if it is empty, so query next group |
| return paths; |
| } |
| } catch (IOException | TException e) { |
| throw new MetadataException(e); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| throw new MetadataException(e); |
| } |
| } |
| logger.warn("Cannot get paths of {} from {}", pathsToQuery, partitionGroup); |
| return Collections.emptyList(); |
| } |
| |
| @SuppressWarnings("java:S1168") // null and empty list are different |
| private List<PartialPath> getMatchedPaths( |
| Node node, Node header, List<String> pathsToQuery, boolean withAlias) |
| throws IOException, TException, InterruptedException { |
| GetAllPathsResult result; |
| if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { |
| AsyncDataClient client = |
| metaGroupMember |
| .getClientProvider() |
| .getAsyncDataClient(node, RaftServer.getReadOperationTimeoutMS()); |
| result = SyncClientAdaptor.getAllPaths(client, header, pathsToQuery, withAlias); |
| } else { |
| try (SyncDataClient syncDataClient = |
| metaGroupMember |
| .getClientProvider() |
| .getSyncDataClient(node, RaftServer.getReadOperationTimeoutMS())) { |
| |
| result = syncDataClient.getAllPaths(header, pathsToQuery, withAlias); |
| } |
| } |
| |
| if (result != null) { |
| // paths may be empty, implying that the group does not contain matched paths, so we do not |
| // need to query other nodes in the group |
| List<PartialPath> partialPaths = new ArrayList<>(); |
| for (int i = 0; i < result.paths.size(); i++) { |
| try { |
| PartialPath partialPath = new PartialPath(result.paths.get(i)); |
| if (withAlias) { |
| partialPath.setMeasurementAlias(result.aliasList.get(i)); |
| } |
| partialPaths.add(partialPath); |
| } catch (IllegalPathException e) { |
| // ignore |
| } |
| } |
| return partialPaths; |
| } else { |
| // a null implies a network failure, so we have to query other nodes in the group |
| return null; |
| } |
| } |
| |
| /** |
| * Split the paths by the data group they belong to and query them from the groups separately. |
| * |
| * @param sgPathMap the key is the storage group name and the value is the path to be queried with |
| * storage group added |
| * @return a collection of all queried devices |
| */ |
| private Set<PartialPath> getMatchedDevices(Map<String, String> sgPathMap) |
| throws MetadataException { |
| Set<PartialPath> result = new HashSet<>(); |
| // split the paths by the data group they belong to |
| Map<PartitionGroup, List<String>> groupPathMap = new HashMap<>(); |
| for (Entry<String, String> sgPathEntry : sgPathMap.entrySet()) { |
| String storageGroupName = sgPathEntry.getKey(); |
| PartialPath pathUnderSG = new PartialPath(sgPathEntry.getValue()); |
| // find the data group that should hold the timeseries schemas of the storage group |
| PartitionGroup partitionGroup = |
| metaGroupMember.getPartitionTable().route(storageGroupName, 0); |
| if (partitionGroup.contains(metaGroupMember.getThisNode())) { |
| // this node is a member of the group, perform a local query after synchronizing with the |
| // leader |
| try { |
| metaGroupMember.getLocalDataMember(partitionGroup.getHeader()).syncLeader(null); |
| } catch (CheckConsistencyException e) { |
| logger.warn("Failed to check consistency.", e); |
| } |
| Set<PartialPath> allDevices = getDevices(pathUnderSG); |
| logger.debug( |
| "{}: get matched paths of {} locally, result {}", |
| metaGroupMember.getName(), |
| partitionGroup, |
| allDevices); |
| result.addAll(allDevices); |
| } else { |
| // batch the queries of the same group to reduce communication |
| groupPathMap |
| .computeIfAbsent(partitionGroup, p -> new ArrayList<>()) |
| .add(pathUnderSG.getFullPath()); |
| } |
| } |
| |
| // query each data group separately |
| for (Entry<PartitionGroup, List<String>> partitionGroupPathEntry : groupPathMap.entrySet()) { |
| PartitionGroup partitionGroup = partitionGroupPathEntry.getKey(); |
| List<String> pathsToQuery = partitionGroupPathEntry.getValue(); |
| |
| result.addAll(getMatchedDevices(partitionGroup, pathsToQuery)); |
| } |
| |
| return result; |
| } |
| |
| private Set<PartialPath> getMatchedDevices( |
| PartitionGroup partitionGroup, List<String> pathsToQuery) throws MetadataException { |
| // choose the node with lowest latency or highest throughput |
| List<Node> coordinatedNodes = QueryCoordinator.getINSTANCE().reorderNodes(partitionGroup); |
| for (Node node : coordinatedNodes) { |
| try { |
| Set<String> paths = getMatchedDevices(node, partitionGroup.getHeader(), pathsToQuery); |
| logger.debug( |
| "{}: get matched paths of {} from {}, result {} for {}", |
| metaGroupMember.getName(), |
| partitionGroup, |
| node, |
| paths, |
| pathsToQuery); |
| if (paths != null) { |
| // query next group |
| Set<PartialPath> partialPaths = new HashSet<>(); |
| for (String path : paths) { |
| partialPaths.add(new PartialPath(path)); |
| } |
| return partialPaths; |
| } |
| } catch (IOException | TException e) { |
| throw new MetadataException(e); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| throw new MetadataException(e); |
| } |
| } |
| logger.warn("Cannot get paths of {} from {}", pathsToQuery, partitionGroup); |
| return Collections.emptySet(); |
| } |
| |
| private Set<String> getMatchedDevices(Node node, Node header, List<String> pathsToQuery) |
| throws IOException, TException, InterruptedException { |
| Set<String> paths; |
| if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { |
| AsyncDataClient client = |
| metaGroupMember |
| .getClientProvider() |
| .getAsyncDataClient(node, RaftServer.getReadOperationTimeoutMS()); |
| paths = SyncClientAdaptor.getAllDevices(client, header, pathsToQuery); |
| } else { |
| try (SyncDataClient syncDataClient = |
| metaGroupMember |
| .getClientProvider() |
| .getSyncDataClient(node, RaftServer.getReadOperationTimeoutMS())) { |
| |
| paths = syncDataClient.getAllDevices(header, pathsToQuery); |
| } |
| } |
| return paths; |
| } |
| |
| /** Similar to method getAllTimeseriesPath(), but return Path with alias alias. */ |
| @Override |
| public Pair<List<PartialPath>, Integer> getAllTimeseriesPathWithAlias( |
| PartialPath prefixPath, int limit, int offset) throws MetadataException { |
| |
| // get all storage groups this path may belong to |
| // the key is the storage group name and the value is the path to be queried with storage group |
| // added, e.g: |
| // "root.*" will be translated into: |
| // "root.group1" -> "root.group1.*", "root.group2" -> "root.group2.*" ... |
| Map<String, String> sgPathMap = determineStorageGroup(prefixPath); |
| List<PartialPath> result = getMatchedPaths(sgPathMap, true); |
| |
| int skippedOffset = 0; |
| // apply offset and limit |
| if (offset > 0 && result.size() > offset) { |
| skippedOffset = offset; |
| result = result.subList(offset, result.size()); |
| } else if (offset > 0) { |
| skippedOffset = result.size(); |
| result = Collections.emptyList(); |
| } |
| if (limit > 0 && result.size() > limit) { |
| result = result.subList(0, limit); |
| } |
| logger.debug("The paths of path {} are {}", prefixPath, result); |
| |
| return new Pair<>(result, skippedOffset); |
| } |
| |
| /** |
| * Get all paths after removing wildcards in the path |
| * |
| * @param originPath a path potentially with wildcard |
| * @return all paths after removing wildcards in the path |
| */ |
| public List<PartialPath> getMatchedPaths(PartialPath originPath) throws MetadataException { |
| // get all storage groups this path may belong to |
| // the key is the storage group name and the value is the path to be queried with storage group |
| // added, e.g: |
| // "root.*" will be translated into: |
| // "root.group1" -> "root.group1.*", "root.group2" -> "root.group2.*" ... |
| Map<String, String> sgPathMap = determineStorageGroup(originPath); |
| List<PartialPath> ret = getMatchedPaths(sgPathMap, false); |
| logger.debug("The paths of path {} are {}", originPath, ret); |
| return ret; |
| } |
| |
| /** |
| * Get all paths after removing wildcards in the path |
| * |
| * @param originalPaths a list of paths, potentially with wildcard |
| * @return a pair of path lists, the first are the existing full paths, the second are invalid |
| * original paths |
| */ |
| public Pair<List<PartialPath>, List<PartialPath>> getMatchedPaths( |
| List<PartialPath> originalPaths) { |
| ConcurrentSkipListSet<PartialPath> fullPaths = new ConcurrentSkipListSet<>(); |
| ConcurrentSkipListSet<PartialPath> nonExistPaths = new ConcurrentSkipListSet<>(); |
| ExecutorService getAllPathsService = |
| Executors.newFixedThreadPool(metaGroupMember.getPartitionTable().getGlobalGroups().size()); |
| for (PartialPath pathStr : originalPaths) { |
| getAllPathsService.submit( |
| () -> { |
| try { |
| List<PartialPath> fullPathStrs = getMatchedPaths(pathStr); |
| if (fullPathStrs.isEmpty()) { |
| nonExistPaths.add(pathStr); |
| logger.debug("Path {} is not found.", pathStr); |
| } else { |
| fullPaths.addAll(fullPathStrs); |
| } |
| } catch (MetadataException e) { |
| logger.error("Failed to get full paths of the prefix path: {} because", pathStr, e); |
| } |
| }); |
| } |
| getAllPathsService.shutdown(); |
| try { |
| getAllPathsService.awaitTermination( |
| RaftServer.getReadOperationTimeoutMS(), TimeUnit.MILLISECONDS); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.error("Unexpected interruption when waiting for get all paths services to stop", e); |
| } |
| return new Pair<>(new ArrayList<>(fullPaths), new ArrayList<>(nonExistPaths)); |
| } |
| |
| /** |
| * Get the local paths that match any path in "paths". The result is not deduplicated. |
| * |
| * @param paths paths potentially contain wildcards |
| */ |
| public List<String> getAllPaths(List<String> paths) throws MetadataException { |
| List<String> ret = new ArrayList<>(); |
| for (String path : paths) { |
| getAllTimeseriesPath(new PartialPath(path)).stream() |
| .map(PartialPath::getFullPath) |
| .forEach(ret::add); |
| } |
| return ret; |
| } |
| |
| /** |
| * Get the local devices that match any path in "paths". The result is deduplicated. |
| * |
| * @param paths paths potentially contain wildcards |
| */ |
| public Set<String> getAllDevices(List<String> paths) throws MetadataException { |
| Set<String> results = new HashSet<>(); |
| for (String path : paths) { |
| getDevices(new PartialPath(path)).stream() |
| .map(PartialPath::getFullPath) |
| .forEach(results::add); |
| } |
| return results; |
| } |
| |
| /** |
| * Get the nodes of a prefix "path" at "nodeLevel". The method currently requires strong |
| * consistency. |
| * |
| * @param path |
| * @param nodeLevel |
| */ |
| public List<String> getNodeList(String path, int nodeLevel) throws MetadataException { |
| return getNodesList(new PartialPath(path), nodeLevel).stream() |
| .map(PartialPath::getFullPath) |
| .collect(Collectors.toList()); |
| } |
| |
| public Set<String> getChildNodeInNextLevel(String path) throws MetadataException { |
| return getChildNodeInNextLevel(new PartialPath(path)); |
| } |
| |
| public Set<String> getChildNodePathInNextLevel(String path) throws MetadataException { |
| return getChildNodePathInNextLevel(new PartialPath(path)); |
| } |
| |
| /** |
| * Replace partial paths (paths not containing measurements), and abstract paths (paths containing |
| * wildcards) with full paths. |
| */ |
| public void convertToFullPaths(PhysicalPlan plan) |
| throws PathNotExistException, CheckConsistencyException { |
| // make sure this node knows all storage groups |
| metaGroupMember.syncLeaderWithConsistencyCheck(false); |
| |
| Pair<List<PartialPath>, List<PartialPath>> getMatchedPathsRet = |
| getMatchedPaths(plan.getPaths()); |
| List<PartialPath> fullPaths = getMatchedPathsRet.left; |
| List<PartialPath> nonExistPath = getMatchedPathsRet.right; |
| plan.setPaths(fullPaths); |
| if (!nonExistPath.isEmpty()) { |
| throw new PathNotExistException( |
| nonExistPath.stream().map(PartialPath::getFullPath).collect(Collectors.toList())); |
| } |
| } |
| |
| @Override |
| public MNode getMNode(MNode deviceMNode, String measurementName) { |
| MNode child = deviceMNode.getChild(measurementName); |
| if (child == null) { |
| child = mRemoteMetaCache.get(deviceMNode.getPartialPath().concatNode(measurementName)); |
| } |
| return child; |
| } |
| |
| public List<ShowTimeSeriesResult> showLocalTimeseries( |
| ShowTimeSeriesPlan plan, QueryContext context) throws MetadataException { |
| return super.showTimeseries(plan, context); |
| } |
| |
| public List<ShowDevicesResult> getLocalDevices(ShowDevicesPlan plan) throws MetadataException { |
| return super.getDevices(plan); |
| } |
| |
| @Override |
| public List<ShowDevicesResult> getDevices(ShowDevicesPlan plan) throws MetadataException { |
| ConcurrentSkipListSet<ShowDevicesResult> resultSet = new ConcurrentSkipListSet<>(); |
| ExecutorService pool = |
| new ThreadPoolExecutor( |
| THREAD_POOL_SIZE, THREAD_POOL_SIZE, 0, TimeUnit.SECONDS, new LinkedBlockingDeque<>()); |
| List<PartitionGroup> globalGroups = metaGroupMember.getPartitionTable().getGlobalGroups(); |
| |
| int limit = plan.getLimit() == 0 ? Integer.MAX_VALUE : plan.getLimit(); |
| int offset = plan.getOffset(); |
| // do not use limit and offset in sub-queries unless offset is 0, otherwise the results are |
| // not combinable |
| if (offset != 0) { |
| plan.setLimit(0); |
| plan.setOffset(0); |
| } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Fetch devices schemas of {} from {} groups", plan.getPath(), globalGroups.size()); |
| } |
| |
| List<Future<Void>> futureList = new ArrayList<>(); |
| for (PartitionGroup group : globalGroups) { |
| futureList.add( |
| pool.submit( |
| () -> { |
| try { |
| getDevices(group, plan, resultSet); |
| } catch (CheckConsistencyException e) { |
| logger.error("Cannot get show devices result of {} from {}", plan, group); |
| } |
| return null; |
| })); |
| } |
| |
| waitForThreadPool(futureList, pool, "getDevices()"); |
| List<ShowDevicesResult> showDevicesResults = |
| applyShowDevicesLimitOffset(resultSet, limit, offset); |
| logger.debug("show devices {} has {} results", plan.getPath(), showDevicesResults.size()); |
| return showDevicesResults; |
| } |
| |
| @Override |
| public List<ShowTimeSeriesResult> showTimeseries(ShowTimeSeriesPlan plan, QueryContext context) |
| throws MetadataException { |
| ConcurrentSkipListSet<ShowTimeSeriesResult> resultSet = new ConcurrentSkipListSet<>(); |
| ExecutorService pool = |
| new ThreadPoolExecutor( |
| THREAD_POOL_SIZE, THREAD_POOL_SIZE, 0, TimeUnit.SECONDS, new LinkedBlockingDeque<>()); |
| List<PartitionGroup> globalGroups = metaGroupMember.getPartitionTable().getGlobalGroups(); |
| |
| int limit = plan.getLimit() == 0 ? Integer.MAX_VALUE : plan.getLimit(); |
| int offset = plan.getOffset(); |
| // do not use limit and offset in sub-queries unless offset is 0, otherwise the results are |
| // not combinable |
| if (offset != 0) { |
| plan.setLimit(0); |
| plan.setOffset(0); |
| } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Fetch timeseries schemas of {} from {} groups", plan.getPath(), globalGroups.size()); |
| } |
| |
| List<Future<Void>> futureList = new ArrayList<>(); |
| for (PartitionGroup group : globalGroups) { |
| futureList.add( |
| pool.submit( |
| () -> { |
| try { |
| showTimeseries(group, plan, resultSet, context); |
| } catch (CheckConsistencyException e) { |
| logger.error("Cannot get show timeseries result of {} from {}", plan, group); |
| } |
| return null; |
| })); |
| } |
| |
| waitForThreadPool(futureList, pool, "showTimeseries()"); |
| List<ShowTimeSeriesResult> showTimeSeriesResults = |
| applyShowTimeseriesLimitOffset(resultSet, limit, offset); |
| logger.debug("Show {} has {} results", plan.getPath(), showTimeSeriesResults.size()); |
| return showTimeSeriesResults; |
| } |
| |
| private List<ShowTimeSeriesResult> applyShowTimeseriesLimitOffset( |
| ConcurrentSkipListSet<ShowTimeSeriesResult> resultSet, int limit, int offset) { |
| List<ShowTimeSeriesResult> showTimeSeriesResults = new ArrayList<>(); |
| Iterator<ShowTimeSeriesResult> iterator = resultSet.iterator(); |
| while (iterator.hasNext() && limit > 0) { |
| if (offset > 0) { |
| offset--; |
| iterator.next(); |
| } else { |
| limit--; |
| showTimeSeriesResults.add(iterator.next()); |
| } |
| } |
| |
| return showTimeSeriesResults; |
| } |
| |
| private List<ShowDevicesResult> applyShowDevicesLimitOffset( |
| Set<ShowDevicesResult> resultSet, int limit, int offset) { |
| List<ShowDevicesResult> showDevicesResults = new ArrayList<>(); |
| Iterator<ShowDevicesResult> iterator = resultSet.iterator(); |
| while (iterator.hasNext() && limit > 0) { |
| if (offset > 0) { |
| offset--; |
| iterator.next(); |
| } else { |
| limit--; |
| showDevicesResults.add(iterator.next()); |
| } |
| } |
| return showDevicesResults; |
| } |
| |
| private void showTimeseries( |
| PartitionGroup group, |
| ShowTimeSeriesPlan plan, |
| Set<ShowTimeSeriesResult> resultSet, |
| QueryContext context) |
| throws CheckConsistencyException, MetadataException { |
| if (group.contains(metaGroupMember.getThisNode())) { |
| showLocalTimeseries(group, plan, resultSet, context); |
| } else { |
| showRemoteTimeseries(group, plan, resultSet); |
| } |
| } |
| |
| private void getDevices( |
| PartitionGroup group, ShowDevicesPlan plan, Set<ShowDevicesResult> resultSet) |
| throws CheckConsistencyException, MetadataException { |
| if (group.contains(metaGroupMember.getThisNode())) { |
| getLocalDevices(group, plan, resultSet); |
| } else { |
| getRemoteDevices(group, plan, resultSet); |
| } |
| } |
| |
| private void getLocalDevices( |
| PartitionGroup group, ShowDevicesPlan plan, Set<ShowDevicesResult> resultSet) |
| throws CheckConsistencyException, MetadataException { |
| Node header = group.getHeader(); |
| DataGroupMember localDataMember = metaGroupMember.getLocalDataMember(header); |
| localDataMember.syncLeaderWithConsistencyCheck(false); |
| try { |
| List<ShowDevicesResult> localResult = super.getDevices(plan); |
| resultSet.addAll(localResult); |
| logger.debug("Fetched {} devices of {} from {}", localResult.size(), plan.getPath(), group); |
| } catch (MetadataException e) { |
| logger.error("Cannot execute show devices plan {} from {} locally.", plan, group); |
| throw e; |
| } |
| } |
| |
| private void showLocalTimeseries( |
| PartitionGroup group, |
| ShowTimeSeriesPlan plan, |
| Set<ShowTimeSeriesResult> resultSet, |
| QueryContext context) |
| throws CheckConsistencyException, MetadataException { |
| Node header = group.getHeader(); |
| DataGroupMember localDataMember = metaGroupMember.getLocalDataMember(header); |
| localDataMember.syncLeaderWithConsistencyCheck(false); |
| try { |
| List<ShowTimeSeriesResult> localResult = super.showTimeseries(plan, context); |
| resultSet.addAll(localResult); |
| logger.debug( |
| "Fetched local timeseries {} schemas of {} from {}", |
| localResult.size(), |
| plan.getPath(), |
| group); |
| } catch (MetadataException e) { |
| logger.error("Cannot execute show timeseries plan {} from {} locally.", plan, group); |
| throw e; |
| } |
| } |
| |
| private void showRemoteTimeseries( |
| PartitionGroup group, ShowTimeSeriesPlan plan, Set<ShowTimeSeriesResult> resultSet) { |
| ByteBuffer resultBinary = null; |
| for (Node node : group) { |
| try { |
| resultBinary = showRemoteTimeseries(node, group, plan); |
| if (resultBinary != null) { |
| break; |
| } |
| } catch (IOException e) { |
| logger.error(LOG_FAIL_CONNECT, node, e); |
| } catch (TException e) { |
| logger.error("Error occurs when getting timeseries schemas in node {}.", node, e); |
| } catch (InterruptedException e) { |
| logger.error("Interrupted when getting timeseries schemas in node {}.", node, e); |
| Thread.currentThread().interrupt(); |
| } |
| } |
| |
| if (resultBinary != null) { |
| int size = resultBinary.getInt(); |
| logger.debug( |
| "Fetched remote timeseries {} schemas of {} from {}", size, plan.getPath(), group); |
| for (int i = 0; i < size; i++) { |
| resultSet.add(ShowTimeSeriesResult.deserialize(resultBinary)); |
| } |
| } else { |
| logger.error("Failed to execute show timeseries {} in group: {}.", plan, group); |
| } |
| } |
| |
| private void getRemoteDevices( |
| PartitionGroup group, ShowDevicesPlan plan, Set<ShowDevicesResult> resultSet) { |
| ByteBuffer resultBinary = null; |
| for (Node node : group) { |
| try { |
| resultBinary = getRemoteDevices(node, group, plan); |
| if (resultBinary != null) { |
| break; |
| } |
| } catch (IOException e) { |
| logger.error(LOG_FAIL_CONNECT, node, e); |
| } catch (TException e) { |
| logger.error("Error occurs when getting devices schemas in node {}.", node, e); |
| } catch (InterruptedException e) { |
| logger.error("Interrupted when getting devices schemas in node {}.", node, e); |
| Thread.currentThread().interrupt(); |
| } |
| } |
| |
| if (resultBinary != null) { |
| int size = resultBinary.getInt(); |
| logger.debug("Fetched remote devices {} schemas of {} from {}", size, plan.getPath(), group); |
| for (int i = 0; i < size; i++) { |
| resultSet.add(ShowDevicesResult.deserialize(resultBinary)); |
| } |
| } else { |
| logger.error("Failed to execute show devices {} in group: {}.", plan, group); |
| } |
| } |
| |
| private ByteBuffer showRemoteTimeseries(Node node, PartitionGroup group, ShowTimeSeriesPlan plan) |
| throws IOException, TException, InterruptedException { |
| ByteBuffer resultBinary; |
| |
| if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { |
| AsyncDataClient client = |
| metaGroupMember |
| .getClientProvider() |
| .getAsyncDataClient(node, RaftServer.getReadOperationTimeoutMS()); |
| resultBinary = SyncClientAdaptor.getAllMeasurementSchema(client, group.getHeader(), plan); |
| } else { |
| try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); |
| DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream); |
| SyncDataClient syncDataClient = |
| metaGroupMember |
| .getClientProvider() |
| .getSyncDataClient(node, RaftServer.getReadOperationTimeoutMS())) { |
| plan.serialize(dataOutputStream); |
| resultBinary = |
| syncDataClient.getAllMeasurementSchema( |
| group.getHeader(), ByteBuffer.wrap(byteArrayOutputStream.toByteArray())); |
| } |
| } |
| return resultBinary; |
| } |
| |
| private ByteBuffer getRemoteDevices(Node node, PartitionGroup group, ShowDevicesPlan plan) |
| throws IOException, TException, InterruptedException { |
| ByteBuffer resultBinary; |
| if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { |
| AsyncDataClient client = |
| metaGroupMember |
| .getClientProvider() |
| .getAsyncDataClient(node, RaftServer.getReadOperationTimeoutMS()); |
| resultBinary = SyncClientAdaptor.getDevices(client, group.getHeader(), plan); |
| } else { |
| try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); |
| DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream); |
| SyncDataClient syncDataClient = |
| metaGroupMember |
| .getClientProvider() |
| .getSyncDataClient(node, RaftServer.getReadOperationTimeoutMS())) { |
| |
| plan.serialize(dataOutputStream); |
| resultBinary = |
| syncDataClient.getDevices( |
| group.getHeader(), ByteBuffer.wrap(byteArrayOutputStream.toByteArray())); |
| } |
| } |
| return resultBinary; |
| } |
| |
| public GetAllPathsResult getAllPaths(List<String> paths, boolean withAlias) |
| throws MetadataException { |
| List<String> retPaths = new ArrayList<>(); |
| List<String> alias = null; |
| if (withAlias) { |
| alias = new ArrayList<>(); |
| } |
| |
| if (withAlias) { |
| for (String path : paths) { |
| List<PartialPath> allTimeseriesPathWithAlias = |
| super.getAllTimeseriesPathWithAlias(new PartialPath(path), -1, -1).left; |
| for (PartialPath timeseriesPathWithAlias : allTimeseriesPathWithAlias) { |
| retPaths.add(timeseriesPathWithAlias.getFullPath()); |
| alias.add(timeseriesPathWithAlias.getMeasurementAlias()); |
| } |
| } |
| } else { |
| retPaths = getAllPaths(paths); |
| } |
| |
| GetAllPathsResult getAllPathsResult = new GetAllPathsResult(); |
| getAllPathsResult.setPaths(retPaths); |
| getAllPathsResult.setAliasList(alias); |
| return getAllPathsResult; |
| } |
| |
| @Override |
| public PartialPath getStorageGroupPath(PartialPath path) throws StorageGroupNotSetException { |
| try { |
| return super.getStorageGroupPath(path); |
| } catch (StorageGroupNotSetException e) { |
| try { |
| metaGroupMember.syncLeader(null); |
| } catch (CheckConsistencyException ex) { |
| logger.warn("Failed to check consistency.", e); |
| } |
| return super.getStorageGroupPath(path); |
| } |
| } |
| } |
