Excluded system & audit from COUNT TIMESERIES and included views (#17703)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBMetadataFetchIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBMetadataFetchIT.java index 1b6818d..82f08cb 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBMetadataFetchIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBMetadataFetchIT.java
@@ -26,6 +26,7 @@ import org.junit.After; import org.junit.Assert; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runners.Parameterized; @@ -495,6 +496,26 @@ } @Test + @Ignore + public void showCountTimeSeriesExcludeInternalDatabaseAndIncludeView() throws SQLException { + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + final long baseVisibleCount = queryCount(statement, "COUNT TIMESERIES root.ln*.**"); + statement.execute("CREATE DATABASE root.count_it"); + statement.execute( + "CREATE TIMESERIES root.count_it.src.s1 WITH DATATYPE = INT32, ENCODING = PLAIN"); + statement.execute( + "CREATE TIMESERIES root.count_it.src.s2 WITH DATATYPE = INT32, ENCODING = PLAIN"); + statement.execute("CREATE VIEW root.count_it.dst.v1 AS SELECT s1 FROM root.count_it.src;"); + + final long localCount = queryCount(statement, "COUNT TIMESERIES root.count_it.**"); + assertEquals(3L, localCount); + assertEquals( + baseVisibleCount + localCount, queryCount(statement, "COUNT TIMESERIES root.**")); + } + } + + @Test public void showCountTimeSeriesWithTag() throws SQLException { try (Connection connection = EnvFactory.getEnv().getConnection(); Statement statement = connection.createStatement()) { @@ -865,4 +886,11 @@ } } } + + private long queryCount(final Statement statement, final String sql) throws SQLException { + try (ResultSet resultSet = statement.executeQuery(sql)) { + Assert.assertTrue(resultSet.next()); + return resultSet.getLong(1); + } + } }
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/regionscan/IoTDBActiveSchemaQueryIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/regionscan/IoTDBActiveSchemaQueryIT.java index c0e7673..dc70614 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/regionscan/IoTDBActiveSchemaQueryIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/regionscan/IoTDBActiveSchemaQueryIT.java
@@ -27,6 +27,7 @@ import org.junit.After; import org.junit.Assert; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runners.Parameterized; @@ -237,6 +238,39 @@ } @Test + @Ignore + public void testCountTimeSeriesWithTimeConditionIncludesView() { + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE root.view_count"); + statement.execute( + "CREATE TIMESERIES root.view_count.src.s1 WITH DATATYPE = INT32, ENCODING = PLAIN"); + statement.execute( + "CREATE TIMESERIES root.view_count.src.s2 WITH DATATYPE = INT32, ENCODING = PLAIN"); + statement.execute("CREATE VIEW root.view_count.dst.v1 AS SELECT s1 FROM root.view_count.src"); + + checkResultSet( + statement, + "count timeseries root.view_count.**", + new HashSet<>(Collections.singletonList("3,"))); + + statement.execute("insert into root.view_count.src(timestamp,s1) values(1,1)"); + + checkResultSet( + statement, + "count timeseries root.view_count.** where time>0", + new HashSet<>(Collections.singletonList("2,"))); + checkResultSet( + statement, + "count timeseries root.view_count.dst.** where time>0", + new HashSet<>(Collections.singletonList("1,"))); + } catch (Exception e) { + e.printStackTrace(); + Assert.fail(e.getMessage()); + } + } + + @Test public void testShowDevices() { try (Connection connection = EnvFactory.getEnv().getConnection(); Statement statement = connection.createStatement()) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/TimeseriesContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/TimeseriesContext.java index 973850e..f2777fa 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/TimeseriesContext.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/TimeseriesContext.java
@@ -30,8 +30,10 @@ import java.nio.ByteBuffer; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; import java.util.Objects; +import java.util.Set; import static org.apache.iotdb.db.queryengine.execution.operator.schema.source.TimeSeriesSchemaSource.mapToString; @@ -47,8 +49,29 @@ private final String deadbandParameters; private final Map<String, String> props; private final String database; + private final int activeCountMultiplier; + private final Set<String> activeLogicalViewCountSet; public TimeseriesContext(IMeasurementSchemaInfo schemaInfo, String database) { + this(schemaInfo, database, 1, Collections.emptySet()); + } + + public TimeseriesContext(IMeasurementSchemaInfo schemaInfo) { + this(schemaInfo, null, 1, Collections.emptySet()); + } + + public TimeseriesContext( + IMeasurementSchemaInfo schemaInfo, + int activeCountMultiplier, + Set<String> activeLogicalViewCountSet) { + this(schemaInfo, null, activeCountMultiplier, activeLogicalViewCountSet); + } + + public TimeseriesContext( + IMeasurementSchemaInfo schemaInfo, + String database, + int activeCountMultiplier, + Set<String> activeLogicalViewCountSet) { this.dataType = schemaInfo.getSchema().getType().toString(); this.encoding = schemaInfo.getSchema().getEncodingType().toString(); this.compression = schemaInfo.getSchema().getCompressor().toString(); @@ -62,6 +85,8 @@ Map<String, String> schemaProps = schemaInfo.getSchema().getProps(); this.props = schemaProps != null ? schemaProps : Collections.emptyMap(); this.database = database; + this.activeCountMultiplier = activeCountMultiplier; + this.activeLogicalViewCountSet = new HashSet<>(activeLogicalViewCountSet); } public String getDataType() { @@ -104,6 +129,64 @@ return database; } + public int getActiveCountMultiplier() { + return activeCountMultiplier; + } + + public Set<String> getActiveLogicalViewCountSet() { + return activeLogicalViewCountSet; + } + + public TimeseriesContext( + String dataType, + String alias, + String encoding, + String compression, + String tags, + String attributes, + String deadband, + String deadbandParameters) { + this( + dataType, + alias, + encoding, + compression, + tags, + attributes, + deadband, + deadbandParameters, + Collections.emptyMap(), + null, + 1, + Collections.emptySet()); + } + + public TimeseriesContext( + String dataType, + String alias, + String encoding, + String compression, + String tags, + String attributes, + String deadband, + String deadbandParameters, + int activeCountMultiplier, + Set<String> activeLogicalViewCountSet) { + this( + dataType, + alias, + encoding, + compression, + tags, + attributes, + deadband, + deadbandParameters, + Collections.emptyMap(), + null, + activeCountMultiplier, + activeLogicalViewCountSet); + } + public TimeseriesContext( String dataType, String alias, @@ -115,6 +198,34 @@ String deadbandParameters, Map<String, String> props, String database) { + this( + dataType, + alias, + encoding, + compression, + tags, + attributes, + deadband, + deadbandParameters, + props, + database, + 1, + Collections.emptySet()); + } + + public TimeseriesContext( + String dataType, + String alias, + String encoding, + String compression, + String tags, + String attributes, + String deadband, + String deadbandParameters, + Map<String, String> props, + String database, + int activeCountMultiplier, + Set<String> activeLogicalViewCountSet) { this.dataType = dataType; this.alias = alias; this.encoding = encoding; @@ -125,6 +236,26 @@ this.deadbandParameters = deadbandParameters; this.props = props != null ? new HashMap<>(props) : new HashMap<>(); this.database = database; + this.activeCountMultiplier = activeCountMultiplier; + this.activeLogicalViewCountSet = new HashSet<>(activeLogicalViewCountSet); + } + + public TimeseriesContext mergeActiveCount(TimeseriesContext that) { + Set<String> mergedActiveLogicalViewCountSet = new HashSet<>(activeLogicalViewCountSet); + mergedActiveLogicalViewCountSet.addAll(that.activeLogicalViewCountSet); + return new TimeseriesContext( + dataType, + alias, + encoding, + compression, + tags, + attributes, + deadband, + deadbandParameters, + props, + database, + activeCountMultiplier + that.activeCountMultiplier, + mergedActiveLogicalViewCountSet); } public void serializeAttributes(ByteBuffer byteBuffer) { @@ -138,6 +269,11 @@ ReadWriteIOUtils.write(deadbandParameters, byteBuffer); ReadWriteIOUtils.write(props, byteBuffer); ReadWriteIOUtils.write(database, byteBuffer); + ReadWriteIOUtils.write(activeCountMultiplier, byteBuffer); + ReadWriteIOUtils.write(activeLogicalViewCountSet.size(), byteBuffer); + for (String logicalView : activeLogicalViewCountSet) { + ReadWriteIOUtils.write(logicalView, byteBuffer); + } } public void serializeAttributes(DataOutputStream stream) throws IOException { @@ -151,6 +287,11 @@ ReadWriteIOUtils.write(deadbandParameters, stream); ReadWriteIOUtils.write(props, stream); ReadWriteIOUtils.write(database, stream); + ReadWriteIOUtils.write(activeCountMultiplier, stream); + ReadWriteIOUtils.write(activeLogicalViewCountSet.size(), stream); + for (String logicalView : activeLogicalViewCountSet) { + ReadWriteIOUtils.write(logicalView, stream); + } } public static TimeseriesContext deserialize(ByteBuffer buffer) { @@ -164,6 +305,12 @@ String deadbandParameters = ReadWriteIOUtils.readString(buffer); Map<String, String> props = ReadWriteIOUtils.readMap(buffer); String database = ReadWriteIOUtils.readString(buffer); + int activeCountMultiplier = ReadWriteIOUtils.readInt(buffer); + int activeLogicalViewCountSetSize = ReadWriteIOUtils.readInt(buffer); + Set<String> activeLogicalViewCountSet = new HashSet<>(); + for (int i = 0; i < activeLogicalViewCountSetSize; i++) { + activeLogicalViewCountSet.add(ReadWriteIOUtils.readString(buffer)); + } return new TimeseriesContext( dataType, alias, @@ -174,7 +321,9 @@ deadband, deadbandParameters, props, - database); + database, + activeCountMultiplier, + activeLogicalViewCountSet); } @Override @@ -196,7 +345,9 @@ && Objects.equals(deadband, that.deadband) && Objects.equals(deadbandParameters, that.deadbandParameters) && Objects.equals(props, that.props) - && Objects.equals(database, that.database); + && Objects.equals(database, that.database) + && activeCountMultiplier == that.activeCountMultiplier + && Objects.equals(activeLogicalViewCountSet, that.activeLogicalViewCountSet); return res; } @@ -212,6 +363,8 @@ deadband, deadbandParameters, props, - database); + database, + activeCountMultiplier, + activeLogicalViewCountSet); } }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/CountGroupByLevelScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/CountGroupByLevelScanOperator.java index 69f3668..35c57da 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/CountGroupByLevelScanOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/CountGroupByLevelScanOperator.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext; import org.apache.iotdb.db.queryengine.execution.operator.schema.source.ISchemaSource; import org.apache.iotdb.db.queryengine.execution.operator.source.SourceOperator; +import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion; import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.ISchemaInfo; import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.reader.ISchemaReader; @@ -95,6 +96,10 @@ return operatorContext; } + private ISchemaRegion getSchemaRegion() { + return ((SchemaDriverContext) operatorContext.getDriverContext()).getSchemaRegion(); + } + @Override public ListenableFuture<?> isBlocked() { if (isBlocked == null) { @@ -109,6 +114,11 @@ */ private ListenableFuture<?> tryGetNext() { if (schemaReader == null) { + if (schemaSource.shouldSkipSchemaRegion(getSchemaRegion())) { + next = null; + isFinished = true; + return NOT_BLOCKED; + } schemaReader = createTimeSeriesReader(); } while (true) { @@ -172,15 +182,14 @@ @Override public boolean hasNext() throws Exception { isBlocked().get(); // wait for the next TsBlock - if (!schemaReader.isSuccess()) { + if (schemaReader != null && !schemaReader.isSuccess()) { throw new SchemaExecutionException(schemaReader.getFailure()); } return next != null; } public ISchemaReader<T> createTimeSeriesReader() { - return schemaSource.getSchemaReader( - ((SchemaDriverContext) operatorContext.getDriverContext()).getSchemaRegion()); + return schemaSource.getSchemaReader(getSchemaRegion()); } private TsBlock constructTsBlockAndClearMap(Map<PartialPath, Long> countMap) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaCountOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaCountOperator.java index 9b28842..4cafe40 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaCountOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaCountOperator.java
@@ -100,6 +100,10 @@ */ private ListenableFuture<?> tryGetNext() { ISchemaRegion schemaRegion = getSchemaRegion(); + if (schemaSource.shouldSkipSchemaRegion(schemaRegion)) { + next = constructTsBlock(0); + return NOT_BLOCKED; + } if (schemaSource.hasSchemaStatistic(schemaRegion)) { long statisticCount = schemaSource.getSchemaStatistic(schemaRegion); // Check if database path itself is counted as a device (bug fix)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/ISchemaSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/ISchemaSource.java index 41bd5f3..4417203 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/ISchemaSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/ISchemaSource.java
@@ -57,6 +57,10 @@ long getSchemaStatistic(final ISchemaRegion schemaRegion); + default boolean shouldSkipSchemaRegion(final ISchemaRegion schemaRegion) { + return false; + } + default boolean checkRegionDatabaseIncluded(final ISchemaRegion schemaRegion) { return true; }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/SchemaSourceFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/SchemaSourceFactory.java index fc4b562..002685e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/SchemaSourceFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/SchemaSourceFactory.java
@@ -49,7 +49,7 @@ Map<Integer, Template> templateMap, PathPatternTree scope) { return new TimeSeriesSchemaSource( - pathPattern, isPrefixMatch, 0, 0, schemaFilter, templateMap, false, scope, true, null); + pathPattern, isPrefixMatch, 0, 0, schemaFilter, templateMap, false, true, scope, true, null); } // show time series @@ -70,6 +70,7 @@ schemaFilter, templateMap, true, + false, scope, true, timeseriesOrdering); @@ -155,6 +156,7 @@ null, // schemaFilter Collections.emptyMap(), // templateMap true, // needViewDetail + false, // excludeInternalDatabase scope, false, // skipInvalidSchema=false to get all series (including invalid) true, // onlyInvalidSchema=true to filter only invalid
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java index 2adb7b6..bed70c0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.commons.schema.column.ColumnHeader; import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; import org.apache.iotdb.commons.schema.filter.SchemaFilter; +import org.apache.iotdb.commons.schema.table.Audit; import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.commons.schema.utils.MeasurementPropsUtils; import org.apache.iotdb.commons.schema.view.ViewType; @@ -56,6 +57,7 @@ private final SchemaFilter schemaFilter; private final Map<Integer, Template> templateMap; private final boolean needViewDetail; + private final boolean excludeInternalDatabase; private final boolean skipInvalidSchema; private final boolean onlyInvalidSchema; private final boolean showInvalidTimeSeries; @@ -69,6 +71,7 @@ SchemaFilter schemaFilter, Map<Integer, Template> templateMap, boolean needViewDetail, + boolean excludeInternalDatabase, PathPatternTree scope, boolean skipInvalidSchema, Ordering timeseriesOrdering) { @@ -80,6 +83,7 @@ schemaFilter, templateMap, needViewDetail, + excludeInternalDatabase, scope, skipInvalidSchema, false, @@ -95,6 +99,7 @@ SchemaFilter schemaFilter, Map<Integer, Template> templateMap, boolean needViewDetail, + boolean excludeInternalDatabase, PathPatternTree scope, boolean skipInvalidSchema, boolean onlyInvalidSchema, @@ -107,6 +112,7 @@ this.schemaFilter = schemaFilter; this.templateMap = templateMap; this.needViewDetail = needViewDetail; + this.excludeInternalDatabase = excludeInternalDatabase; this.scope = scope; this.skipInvalidSchema = skipInvalidSchema; this.onlyInvalidSchema = onlyInvalidSchema; @@ -194,6 +200,25 @@ return schemaRegion.getSchemaRegionStatistics().getSeriesNumber(true, false); } + @Override + public boolean shouldSkipSchemaRegion(final ISchemaRegion schemaRegion) { + if (!excludeInternalDatabase) { + return false; + } + + final String database = schemaRegion.getDatabaseFullPath(); + if (!SchemaConstant.SYSTEM_DATABASE.equals(database) + && !SchemaConstant.AUDIT_DATABASE.equals(database) + && !Audit.TABLE_MODEL_AUDIT_DATABASE.equals(database)) { + return false; + } + + final String[] nodes = pathPattern.getNodes(); + return nodes.length < 2 + || !SchemaConstant.ROOT.equals(nodes[0]) + || !database.endsWith("." + nodes[1]); + } + public static String mapToString(Map<String, String> map) { if (map == null || map.isEmpty()) { return null;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/ActiveTimeSeriesRegionScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/ActiveTimeSeriesRegionScanOperator.java index 5664e43..ddd5409 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/ActiveTimeSeriesRegionScanOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/ActiveTimeSeriesRegionScanOperator.java
@@ -36,13 +36,16 @@ import org.apache.tsfile.utils.RamUsageEstimator; import java.io.IOException; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.stream.Collectors; public class ActiveTimeSeriesRegionScanOperator extends AbstractRegionScanDataSourceOperator { // Timeseries which need to be checked. private final Map<IDeviceID, Map<String, TimeseriesContext>> timeSeriesToSchemasInfo; + private final Set<String> countedLogicalViews; private static final Binary VIEW_TYPE = new Binary("BASE".getBytes()); private final Binary dataBaseName; private final boolean onlyInvalidSchema; @@ -63,6 +66,7 @@ this.operatorContext = operatorContext; this.sourceId = sourceId; this.timeSeriesToSchemasInfo = timeSeriesToSchemasInfo; + this.countedLogicalViews = new HashSet<>(); this.regionScanUtil = new RegionScanForActiveTimeSeriesUtil(timeFilter, ttlCache); this.onlyInvalidSchema = onlyInvalidSchema; this.dataBaseName = @@ -105,7 +109,16 @@ if (outputCount) { for (Map.Entry<IDeviceID, List<String>> entry : activeTimeSeries.entrySet()) { List<String> timeSeriesList = entry.getValue(); - count += timeSeriesList.size(); + Map<String, TimeseriesContext> timeSeriesInfo = timeSeriesToSchemasInfo.get(entry.getKey()); + for (String timeSeries : timeSeriesList) { + TimeseriesContext schemaInfo = timeSeriesInfo.get(timeSeries); + count += schemaInfo.getActiveCountMultiplier(); + for (String logicalView : schemaInfo.getActiveLogicalViewCountSet()) { + if (countedLogicalViews.add(logicalView)) { + count++; + } + } + } removeTimeseriesListFromDevice(entry.getKey(), timeSeriesList); } return;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java index 32e26a9..d09dc68 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
@@ -3338,7 +3338,8 @@ MPPQueryContext context, PathPatternTree authorityScope, boolean canSeeAuditDB, - boolean isOnlyNeedInvalidTimeSeries) + boolean isOnlyNeedInvalidTimeSeries, + boolean includeLogicalView) throws IllegalPathException { analyzeGlobalTimeConditionInShowMetaData(timeCondition, analysis); context.generateGlobalTimeFilter(analysis); @@ -3354,29 +3355,50 @@ analysis.setFinishQueryAfterAnalyze(true); return false; } - removeLogicViewMeasurement(schemaTree); - Map<PartialPath, Map<PartialPath, List<TimeseriesContext>>> deviceToTimeseriesContext; + + List<DeviceSchemaInfo> deviceSchemaInfoList; + if (includeLogicalView) { + deviceSchemaInfoList = schemaTree.getMatchedDevices(ALL_MATCH_PATTERN); + updateSchemaTreeByViews(analysis, schemaTree, context, canSeeAuditDB); + } else { + removeLogicViewMeasurement(schemaTree); + deviceSchemaInfoList = schemaTree.getMatchedDevices(ALL_MATCH_PATTERN); + } + + Map<PartialPath, Map<PartialPath, List<TimeseriesContext>>> deviceToTimeseriesContext = + new HashMap<>(); /** * Since we fetch raw time series schema without template(The template sequence will be treated * as a normal node, not a device+templateId. This means that all nodes are what we need.). We * can use ALL_MATCH_PATTERN to get result. */ - List<DeviceSchemaInfo> deviceSchemaInfoList = schemaTree.getMatchedDevices(ALL_MATCH_PATTERN); - Set<String> databases = schemaTree.getDatabases(); - - deviceToTimeseriesContext = - buildDeviceToTimeseriesContext( - deviceSchemaInfoList, patternTree, databases, false, isOnlyNeedInvalidTimeSeries); + Set<IDeviceID> deviceSet = new HashSet<>(); + if (includeLogicalView) { + deviceSet = + buildDeviceToTimeseriesContextForActiveCount( + deviceSchemaInfoList, + patternTree, + schemaTree.getDatabases(), + schemaTree, + deviceToTimeseriesContext, + isOnlyNeedInvalidTimeSeries); + } else { + deviceToTimeseriesContext = + buildDeviceToTimeseriesContext( + deviceSchemaInfoList, + patternTree, + schemaTree.getDatabases(), + false, + isOnlyNeedInvalidTimeSeries); + deviceSet = + deviceToTimeseriesContext.keySet().stream() + .map(PartialPath::getIDeviceIDAsFullDevice) + .collect(Collectors.toSet()); + } analysis.setDeviceToTimeseriesSchemas(deviceToTimeseriesContext); // fetch Data partition - DataPartition dataPartition = - fetchDataPartitionByDevices( - deviceToTimeseriesContext.keySet().stream() - .map(PartialPath::getIDeviceIDAsFullDevice) - .collect(Collectors.toSet()), - schemaTree, - context); + DataPartition dataPartition = fetchDataPartitionByDevices(deviceSet, schemaTree, context); analysis.setDataPartitionInfo(dataPartition); return true; } @@ -3423,6 +3445,169 @@ return null; } + private Set<IDeviceID> buildDeviceToTimeseriesContextForActiveCount( + List<DeviceSchemaInfo> deviceSchemaInfoList, + PathPatternTree patternTree, + Set<String> databases, + ISchemaTree schemaTree, + Map<PartialPath, Map<PartialPath, List<TimeseriesContext>>> deviceToTimeseriesContext, + boolean skipAliasSeries) { + Set<IDeviceID> deviceSet = new HashSet<>(); + for (DeviceSchemaInfo deviceSchemaInfo : deviceSchemaInfoList) { + PartialPath devicePath = deviceSchemaInfo.getDevicePath(); + List<IMeasurementSchemaInfo> physicalMeasurementSchemaInfoList = new ArrayList<>(); + List<IMeasurementSchemaInfo> logicalViewSchemaInfoList = new ArrayList<>(); + for (IMeasurementSchemaInfo measurementSchemaInfo : + deviceSchemaInfo.getMeasurementSchemaInfoList()) { + if (measurementSchemaInfo.isLogicalView()) { + logicalViewSchemaInfoList.add(measurementSchemaInfo); + } else { + physicalMeasurementSchemaInfoList.add(measurementSchemaInfo); + } + } + + Pair<List<IMeasurementSchemaInfo>, List<String>> validMeasurements = + collectValidMeasurements( + physicalMeasurementSchemaInfoList, + devicePath, + patternTree, + databases, + false, + skipAliasSeries); + if (!validMeasurements.left.isEmpty()) { + if (deviceSchemaInfo.isAligned()) { + addAlignedDeviceToActiveCountContext( + deviceToTimeseriesContext, + devicePath, + validMeasurements.left, + validMeasurements.right, + deviceSet); + } else { + addNonAlignedDeviceToActiveCountContext( + deviceToTimeseriesContext, + devicePath, + validMeasurements.left, + validMeasurements.right, + deviceSet); + } + } + + for (IMeasurementSchemaInfo logicalViewSchemaInfo : logicalViewSchemaInfoList) { + addLogicalViewSourcesForActiveCount( + devicePath, + logicalViewSchemaInfo, + schemaTree, + databases, + deviceToTimeseriesContext, + deviceSet); + } + } + return deviceSet; + } + + private void addAlignedDeviceToActiveCountContext( + Map<PartialPath, Map<PartialPath, List<TimeseriesContext>>> deviceToTimeseriesContext, + PartialPath devicePath, + List<IMeasurementSchemaInfo> validMeasurementSchemaInfoList, + List<String> validDatabaseList, + Set<IDeviceID> deviceSet) { + deviceSet.add(devicePath.getIDeviceIDAsFullDevice()); + addAlignedDeviceToContext( + deviceToTimeseriesContext, devicePath, validMeasurementSchemaInfoList, validDatabaseList); + } + + private void addNonAlignedDeviceToActiveCountContext( + Map<PartialPath, Map<PartialPath, List<TimeseriesContext>>> deviceToTimeseriesContext, + PartialPath devicePath, + List<IMeasurementSchemaInfo> validMeasurementSchemaInfoList, + List<String> validDatabaseList, + Set<IDeviceID> deviceSet) { + for (int i = 0; i < validMeasurementSchemaInfoList.size(); i++) { + IMeasurementSchemaInfo measurementSchemaInfo = validMeasurementSchemaInfoList.get(i); + addPhysicalTimeseriesForActiveCount( + devicePath, + measurementSchemaInfo, + false, + new TimeseriesContext(measurementSchemaInfo, validDatabaseList.get(i)), + deviceToTimeseriesContext, + deviceSet); + } + } + + private void addLogicalViewSourcesForActiveCount( + PartialPath viewDevicePath, + IMeasurementSchemaInfo viewSchemaInfo, + ISchemaTree schemaTree, + Set<String> databases, + Map<PartialPath, Map<PartialPath, List<TimeseriesContext>>> deviceToTimeseriesContext, + Set<IDeviceID> deviceSet) { + LogicalViewSchema logicalViewSchema = viewSchemaInfo.getSchemaAsLogicalViewSchema(); + if (logicalViewSchema == null) { + return; + } + + String viewPath = viewDevicePath.concatNode(viewSchemaInfo.getName()).getFullPath(); + for (PartialPath sourcePath : getSourcePaths(logicalViewSchema.getExpression())) { + if (sourcePath.getNodeLength() <= 1) { + continue; + } + PartialPath sourceDevicePath = + new PartialPath(Arrays.copyOf(sourcePath.getNodes(), sourcePath.getNodeLength() - 1)); + DeviceSchemaInfo sourceDeviceSchemaInfo = + schemaTree.searchDeviceSchemaInfo( + sourceDevicePath, Collections.singletonList(sourcePath.getMeasurement())); + if (sourceDeviceSchemaInfo == null + || sourceDeviceSchemaInfo.getMeasurementSchemaInfoList().isEmpty()) { + continue; + } + + IMeasurementSchemaInfo sourceSchemaInfo = + sourceDeviceSchemaInfo.getMeasurementSchemaInfoList().get(0); + if (sourceSchemaInfo == null || sourceSchemaInfo.isLogicalView()) { + continue; + } + + addPhysicalTimeseriesForActiveCount( + sourceDevicePath, + sourceSchemaInfo, + sourceDeviceSchemaInfo.isAligned(), + new TimeseriesContext( + sourceSchemaInfo, + getDatabaseForMeasurement(sourceSchemaInfo, sourceDevicePath, databases, false), + 0, + Collections.singleton(viewPath)), + deviceToTimeseriesContext, + deviceSet); + } + } + + private void addPhysicalTimeseriesForActiveCount( + PartialPath devicePath, + IMeasurementSchemaInfo measurementSchemaInfo, + boolean isAligned, + TimeseriesContext timeseriesContext, + Map<PartialPath, Map<PartialPath, List<TimeseriesContext>>> deviceToTimeseriesContext, + Set<IDeviceID> deviceSet) { + deviceSet.add(devicePath.getIDeviceIDAsFullDevice()); + PartialPath timeseriesPath = + isAligned + ? new AlignedPath( + devicePath.getNodes(), + Collections.singletonList(measurementSchemaInfo.getName()), + Collections.singletonList(measurementSchemaInfo.getSchema())) + : new MeasurementPath( + devicePath.concatNode(measurementSchemaInfo.getName()).getNodes()); + Map<PartialPath, List<TimeseriesContext>> timeseriesContextMap = + deviceToTimeseriesContext.computeIfAbsent(devicePath, k -> new HashMap<>()); + List<TimeseriesContext> existingContextList = timeseriesContextMap.get(timeseriesPath); + if (existingContextList == null) { + timeseriesContextMap.put( + timeseriesPath, new ArrayList<>(Collections.singletonList(timeseriesContext))); + } else { + existingContextList.set(0, existingContextList.get(0).mergeActiveCount(timeseriesContext)); + } + } + @Override public Analysis visitShowTimeSeries( ShowTimeSeriesStatement showTimeSeriesStatement, MPPQueryContext context) { @@ -3453,7 +3638,8 @@ context, showTimeSeriesStatement.getAuthorityScope(), showTimeSeriesStatement.isCanSeeAuditDB(), - showTimeSeriesStatement.isOnlyShowDisable()); + showTimeSeriesStatement.isOnlyShowDisable(), + false); if (!hasSchema) { if (showTimeSeriesStatement.isOnlyShowDisable()) { analysis.setRespDatasetHeader(DatasetHeaderFactory.getShowInvalidTimeSeriesHeader()); @@ -3957,7 +4143,8 @@ context, countTimeSeriesStatement.getAuthorityScope(), countTimeSeriesStatement.isCanSeeAuditDB(), - false); + false, + true); if (!hasSchema) { analysis.setRespDatasetHeader(DatasetHeaderFactory.getCountTimeSeriesHeader()); return analysis;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java index ce3a9fe..79427c8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
@@ -3803,13 +3803,20 @@ for (Map.Entry<PartialPath, List<TimeseriesContext>> entry : entryMap.getValue().entrySet()) { PartialPath path = entry.getKey(); if (path instanceof MeasurementPath) { - timeseriesSchemaInfoMap.put(path.getMeasurement(), entry.getValue().get(0)); - context.addPath( - new NonAlignedFullPath( - path.getIDeviceID(), - new MeasurementSchema( - path.getMeasurement(), - TSDataType.valueOf(entry.getValue().get(0).getDataType())))); + String measurement = path.getMeasurement(); + TimeseriesContext timeseriesContext = entry.getValue().get(0); + TimeseriesContext existingContext = timeseriesSchemaInfoMap.get(measurement); + if (existingContext == null) { + timeseriesSchemaInfoMap.put(measurement, timeseriesContext); + context.addPath( + new NonAlignedFullPath( + path.getIDeviceID(), + new MeasurementSchema( + measurement, TSDataType.valueOf(timeseriesContext.getDataType())))); + } else { + timeseriesSchemaInfoMap.put( + measurement, existingContext.mergeActiveCount(timeseriesContext)); + } } else if (path instanceof AlignedPath) { AlignedPath alignedPath = (AlignedPath) path; List<String> measurementList = alignedPath.getMeasurementList(); @@ -3819,14 +3826,25 @@ } int size = measurementList.size(); List<IMeasurementSchema> schemaList = new ArrayList<>(size); + List<String> newMeasurementList = new ArrayList<>(size); for (int i = 0; i < size; i++) { - timeseriesSchemaInfoMap.put(measurementList.get(i), entry.getValue().get(i)); - schemaList.add( - new MeasurementSchema( - measurementList.get(i), - TSDataType.valueOf(entry.getValue().get(i).getDataType()))); + String measurement = measurementList.get(i); + TimeseriesContext timeseriesContext = entry.getValue().get(i); + TimeseriesContext existingContext = timeseriesSchemaInfoMap.get(measurement); + if (existingContext == null) { + timeseriesSchemaInfoMap.put(measurement, timeseriesContext); + newMeasurementList.add(measurement); + schemaList.add( + new MeasurementSchema( + measurement, TSDataType.valueOf(timeseriesContext.getDataType()))); + } else { + timeseriesSchemaInfoMap.put( + measurement, existingContext.mergeActiveCount(timeseriesContext)); + } } - context.addPath(new AlignedFullPath(path.getIDeviceID(), measurementList, schemaList)); + if (!newMeasurementList.isEmpty()) { + context.addPath(new AlignedFullPath(path.getIDeviceID(), newMeasurementList, schemaList)); + } } } return timeseriesSchemaInfoMap;
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaCountOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaCountOperatorTest.java index b59689a..3cd50eb 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaCountOperatorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaCountOperatorTest.java
@@ -115,6 +115,82 @@ } @Test + public void testSchemaCountOperatorSkipSchemaRegion() throws Exception { + ExecutorService instanceNotificationExecutor = + IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification"); + try { + QueryId queryId = new QueryId("stub_query"); + FragmentInstanceId instanceId = + new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance"); + FragmentInstanceStateMachine stateMachine = + new FragmentInstanceStateMachine(instanceId, instanceNotificationExecutor); + FragmentInstanceContext fragmentInstanceContext = + createFragmentInstanceContext(instanceId, stateMachine); + DriverContext driverContext = new DriverContext(fragmentInstanceContext, 0); + PlanNodeId planNodeId = queryId.genPlanNodeId(); + ISchemaRegion schemaRegion = Mockito.mock(ISchemaRegion.class); + OperatorContext operatorContext = + driverContext.addOperatorContext( + 1, planNodeId, SchemaCountOperator.class.getSimpleName()); + operatorContext.setDriverContext( + new SchemaDriverContext(fragmentInstanceContext, schemaRegion, 0)); + ISchemaSource<ISchemaInfo> schemaSource = Mockito.mock(ISchemaSource.class); + Mockito.when(schemaSource.shouldSkipSchemaRegion(schemaRegion)).thenReturn(true); + + SchemaCountOperator<?> schemaCountOperator = + new SchemaCountOperator<>( + planNodeId, driverContext.getOperatorContexts().get(0), schemaSource); + + assertTrue(schemaCountOperator.hasNext()); + TsBlock tsBlock = schemaCountOperator.next(); + assertEquals(0, tsBlock.getColumn(0).getLong(0)); + assertTrue(schemaCountOperator.isFinished()); + Mockito.verify(schemaSource, Mockito.never()).getSchemaReader(schemaRegion); + } finally { + instanceNotificationExecutor.shutdown(); + } + } + + @Test + public void testSchemaCountOperatorUseSchemaStatistic() throws Exception { + ExecutorService instanceNotificationExecutor = + IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification"); + try { + QueryId queryId = new QueryId("stub_query"); + FragmentInstanceId instanceId = + new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance"); + FragmentInstanceStateMachine stateMachine = + new FragmentInstanceStateMachine(instanceId, instanceNotificationExecutor); + FragmentInstanceContext fragmentInstanceContext = + createFragmentInstanceContext(instanceId, stateMachine); + DriverContext driverContext = new DriverContext(fragmentInstanceContext, 0); + PlanNodeId planNodeId = queryId.genPlanNodeId(); + ISchemaRegion schemaRegion = Mockito.mock(ISchemaRegion.class); + OperatorContext operatorContext = + driverContext.addOperatorContext( + 1, planNodeId, SchemaCountOperator.class.getSimpleName()); + operatorContext.setDriverContext( + new SchemaDriverContext(fragmentInstanceContext, schemaRegion, 0)); + ISchemaSource<ISchemaInfo> schemaSource = Mockito.mock(ISchemaSource.class); + Mockito.when(schemaSource.hasSchemaStatistic(schemaRegion)).thenReturn(true); + Mockito.when(schemaSource.getSchemaStatistic(schemaRegion)).thenReturn(7L); + Mockito.when(schemaSource.checkRegionDatabaseIncluded(schemaRegion)).thenReturn(true); + + SchemaCountOperator<?> schemaCountOperator = + new SchemaCountOperator<>( + planNodeId, driverContext.getOperatorContexts().get(0), schemaSource); + + assertTrue(schemaCountOperator.hasNext()); + TsBlock tsBlock = schemaCountOperator.next(); + assertEquals(7, tsBlock.getColumn(0).getLong(0)); + Mockito.verify(schemaSource).getSchemaStatistic(schemaRegion); + Mockito.verify(schemaSource, Mockito.never()).getSchemaReader(schemaRegion); + } finally { + instanceNotificationExecutor.shutdown(); + } + } + + @Test public void testLevelTimeSeriesCountOperator() { ExecutorService instanceNotificationExecutor = IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification"); @@ -185,6 +261,43 @@ } } + @Test + public void testLevelTimeSeriesCountOperatorSkipSchemaRegion() { + ExecutorService instanceNotificationExecutor = + IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification"); + try { + QueryId queryId = new QueryId("stub_query"); + FragmentInstanceId instanceId = + new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance"); + FragmentInstanceStateMachine stateMachine = + new FragmentInstanceStateMachine(instanceId, instanceNotificationExecutor); + FragmentInstanceContext fragmentInstanceContext = + createFragmentInstanceContext(instanceId, stateMachine); + DriverContext driverContext = new DriverContext(fragmentInstanceContext, 0); + PlanNodeId planNodeId = queryId.genPlanNodeId(); + OperatorContext operatorContext = + driverContext.addOperatorContext( + 1, planNodeId, CountGroupByLevelScanOperator.class.getSimpleName()); + ISchemaRegion schemaRegion = Mockito.mock(ISchemaRegion.class); + operatorContext.setDriverContext( + new SchemaDriverContext(fragmentInstanceContext, schemaRegion, 0)); + ISchemaSource<ITimeSeriesSchemaInfo> schemaSource = Mockito.mock(ISchemaSource.class); + Mockito.when(schemaSource.shouldSkipSchemaRegion(schemaRegion)).thenReturn(true); + + CountGroupByLevelScanOperator<ITimeSeriesSchemaInfo> timeSeriesCountOperator = + new CountGroupByLevelScanOperator<>( + planNodeId, driverContext.getOperatorContexts().get(0), 1, schemaSource); + + assertTrue(collectResult(timeSeriesCountOperator).isEmpty()); + Mockito.verify(schemaSource, Mockito.never()).getSchemaReader(schemaRegion); + } catch (Exception e) { + e.printStackTrace(); + fail(); + } finally { + instanceNotificationExecutor.shutdown(); + } + } + private List<TsBlock> collectResult(CountGroupByLevelScanOperator<?> operator) throws Exception { List<TsBlock> tsBlocks = new ArrayList<>(); while (operator.hasNext()) {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSourceTest.java new file mode 100644 index 0000000..a842070 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSourceTest.java
@@ -0,0 +1,168 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.queryengine.execution.operator.schema.source; + +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.schema.SchemaConstant; +import org.apache.iotdb.commons.schema.table.Audit; +import org.apache.iotdb.db.schemaengine.rescon.ISchemaRegionStatistics; +import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion; +import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.ITimeSeriesSchemaInfo; + +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.Collections; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class TimeSeriesSchemaSourceTest { + + @Test + public void testCountSourceSkipsImplicitInternalDatabases() throws Exception { + final ISchemaSource<ITimeSeriesSchemaInfo> countSource = + SchemaSourceFactory.getTimeSeriesSchemaCountSource( + new PartialPath("root.**"), + false, + null, + Collections.emptyMap(), + SchemaConstant.ALL_MATCH_SCOPE); + + assertTrue( + countSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.SYSTEM_DATABASE))); + assertTrue(countSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.AUDIT_DATABASE))); + assertTrue( + countSource.shouldSkipSchemaRegion(mockSchemaRegion(Audit.TABLE_MODEL_AUDIT_DATABASE))); + assertFalse(countSource.shouldSkipSchemaRegion(mockSchemaRegion("root.sg"))); + } + + @Test + public void testCountSourceKeepsExplicitInternalDatabaseQueries() throws Exception { + final ISchemaSource<ITimeSeriesSchemaInfo> systemCountSource = + SchemaSourceFactory.getTimeSeriesSchemaCountSource( + new PartialPath("root.__system.**"), + false, + null, + Collections.emptyMap(), + SchemaConstant.ALL_MATCH_SCOPE); + assertFalse( + systemCountSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.SYSTEM_DATABASE))); + assertTrue( + systemCountSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.AUDIT_DATABASE))); + + final ISchemaSource<ITimeSeriesSchemaInfo> auditCountSource = + SchemaSourceFactory.getTimeSeriesSchemaCountSource( + new PartialPath("root.__audit.**"), + false, + null, + Collections.emptyMap(), + SchemaConstant.ALL_MATCH_SCOPE); + assertFalse( + auditCountSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.AUDIT_DATABASE))); + assertTrue( + auditCountSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.SYSTEM_DATABASE))); + } + + @Test + public void testCountSourceSkipsWildcardSecondNodeForInternalDatabases() throws Exception { + final ISchemaSource<ITimeSeriesSchemaInfo> countSource = + SchemaSourceFactory.getTimeSeriesSchemaCountSource( + new PartialPath("root.*.**"), + false, + null, + Collections.emptyMap(), + SchemaConstant.ALL_MATCH_SCOPE); + + assertTrue( + countSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.SYSTEM_DATABASE))); + assertTrue(countSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.AUDIT_DATABASE))); + assertFalse(countSource.shouldSkipSchemaRegion(mockSchemaRegion("root.sg"))); + } + + @Test + public void testCountSourceKeepsExactInternalDatabaseQueries() throws Exception { + final ISchemaSource<ITimeSeriesSchemaInfo> systemCountSource = + SchemaSourceFactory.getTimeSeriesSchemaCountSource( + new PartialPath("root.__system"), + false, + null, + Collections.emptyMap(), + SchemaConstant.ALL_MATCH_SCOPE); + assertFalse( + systemCountSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.SYSTEM_DATABASE))); + + final ISchemaSource<ITimeSeriesSchemaInfo> auditCountSource = + SchemaSourceFactory.getTimeSeriesSchemaCountSource( + new PartialPath("root.__audit"), + false, + null, + Collections.emptyMap(), + SchemaConstant.ALL_MATCH_SCOPE); + assertFalse( + auditCountSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.AUDIT_DATABASE))); + } + + @Test + public void testShowSourceDoesNotSkipInternalDatabases() throws Exception { + final ISchemaSource<ITimeSeriesSchemaInfo> showSource = + SchemaSourceFactory.getTimeSeriesSchemaScanSource( + new PartialPath("root.**"), + false, + 0, + 0, + null, + Collections.emptyMap(), + SchemaConstant.ALL_MATCH_SCOPE, + null); + + assertFalse( + showSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.SYSTEM_DATABASE))); + assertFalse(showSource.shouldSkipSchemaRegion(mockSchemaRegion(SchemaConstant.AUDIT_DATABASE))); + } + + @Test + public void testCountStatisticIncludesView() throws Exception { + final ISchemaSource<ITimeSeriesSchemaInfo> countSource = + SchemaSourceFactory.getTimeSeriesSchemaCountSource( + new PartialPath("root.sg.**"), + false, + null, + Collections.emptyMap(), + SchemaConstant.ALL_MATCH_SCOPE); + final ISchemaRegion schemaRegion = mockSchemaRegion("root.sg"); + final ISchemaRegionStatistics schemaRegionStatistics = + Mockito.mock(ISchemaRegionStatistics.class); + + Mockito.when(schemaRegion.getSchemaRegionStatistics()).thenReturn(schemaRegionStatistics); + Mockito.when(schemaRegionStatistics.getSeriesNumber(true, false)).thenReturn(5L); + + assertEquals(5L, countSource.getSchemaStatistic(schemaRegion)); + Mockito.verify(schemaRegionStatistics).getSeriesNumber(true, false); + Mockito.verify(schemaRegionStatistics, Mockito.never()).getSeriesNumber(false, false); + } + + private ISchemaRegion mockSchemaRegion(final String database) { + final ISchemaRegion schemaRegion = Mockito.mock(ISchemaRegion.class); + Mockito.when(schemaRegion.getDatabaseFullPath()).thenReturn(database); + return schemaRegion; + } +}