Skip to content

Commit 3c959c3

Browse files
CaideyipiCaideyipi
authored
Excluded system & audit from COUNT TIMESERIES and included views (#17703) (#17792)
(cherry picked from commit 312d384)
1 parent e94f0e8 commit 3c959c3

13 files changed

Lines changed: 626 additions & 32 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBMetadataFetchIT.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.junit.After;
2727
import org.junit.Assert;
2828
import org.junit.Before;
29+
import org.junit.Ignore;
2930
import org.junit.Test;
3031
import org.junit.experimental.categories.Category;
3132
import org.junit.runners.Parameterized;
@@ -494,6 +495,26 @@ public void showCountTimeSeries() throws SQLException {
494495
}
495496
}
496497

498+
@Test
499+
@Ignore
500+
public void showCountTimeSeriesExcludeInternalDatabaseAndIncludeView() throws SQLException {
501+
try (Connection connection = EnvFactory.getEnv().getConnection();
502+
Statement statement = connection.createStatement()) {
503+
final long baseVisibleCount = queryCount(statement, "COUNT TIMESERIES root.ln*.**");
504+
statement.execute("CREATE DATABASE root.count_it");
505+
statement.execute(
506+
"CREATE TIMESERIES root.count_it.src.s1 WITH DATATYPE = INT32, ENCODING = PLAIN");
507+
statement.execute(
508+
"CREATE TIMESERIES root.count_it.src.s2 WITH DATATYPE = INT32, ENCODING = PLAIN");
509+
statement.execute("CREATE VIEW root.count_it.dst.v1 AS SELECT s1 FROM root.count_it.src;");
510+
511+
final long localCount = queryCount(statement, "COUNT TIMESERIES root.count_it.**");
512+
assertEquals(3L, localCount);
513+
assertEquals(
514+
baseVisibleCount + localCount, queryCount(statement, "COUNT TIMESERIES root.**"));
515+
}
516+
}
517+
497518
@Test
498519
public void showCountTimeSeriesWithTag() throws SQLException {
499520
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -865,4 +886,11 @@ public void showDeadbandInfo() throws SQLException {
865886
}
866887
}
867888
}
889+
890+
private long queryCount(final Statement statement, final String sql) throws SQLException {
891+
try (ResultSet resultSet = statement.executeQuery(sql)) {
892+
Assert.assertTrue(resultSet.next());
893+
return resultSet.getLong(1);
894+
}
895+
}
868896
}

‎integration-test/src/test/java/org/apache/iotdb/db/it/schema/regionscan/IoTDBActiveSchemaQueryIT.java‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.junit.After;
2828
import org.junit.Assert;
2929
import org.junit.Before;
30+
import org.junit.Ignore;
3031
import org.junit.Test;
3132
import org.junit.experimental.categories.Category;
3233
import org.junit.runners.Parameterized;
@@ -235,6 +236,39 @@ public void testShowTimeSeries() {
235236
}
236237
}
237238

239+
@Test
240+
@Ignore
241+
public void testCountTimeSeriesWithTimeConditionIncludesView() {
242+
try (Connection connection = EnvFactory.getEnv().getConnection();
243+
Statement statement = connection.createStatement()) {
244+
statement.execute("CREATE DATABASE root.view_count");
245+
statement.execute(
246+
"CREATE TIMESERIES root.view_count.src.s1 WITH DATATYPE = INT32, ENCODING = PLAIN");
247+
statement.execute(
248+
"CREATE TIMESERIES root.view_count.src.s2 WITH DATATYPE = INT32, ENCODING = PLAIN");
249+
statement.execute("CREATE VIEW root.view_count.dst.v1 AS SELECT s1 FROM root.view_count.src");
250+
251+
checkResultSet(
252+
statement,
253+
"count timeseries root.view_count.**",
254+
new HashSet<>(Collections.singletonList("3,")));
255+
256+
statement.execute("insert into root.view_count.src(timestamp,s1) values(1,1)");
257+
258+
checkResultSet(
259+
statement,
260+
"count timeseries root.view_count.** where time>0",
261+
new HashSet<>(Collections.singletonList("2,")));
262+
checkResultSet(
263+
statement,
264+
"count timeseries root.view_count.dst.** where time>0",
265+
new HashSet<>(Collections.singletonList("1,")));
266+
} catch (Exception e) {
267+
e.printStackTrace();
268+
Assert.fail(e.getMessage());
269+
}
270+
}
271+
238272
@Test
239273
public void testShowDevices() {
240274
try (Connection connection = EnvFactory.getEnv().getConnection();

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/TimeseriesContext.java‎

Lines changed: 103 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,10 @@
2828
import java.io.DataOutputStream;
2929
import java.io.IOException;
3030
import java.nio.ByteBuffer;
31+
import java.util.Collections;
32+
import java.util.HashSet;
3133
import java.util.Objects;
34+
import java.util.Set;
3235

3336
import static org.apache.iotdb.db.queryengine.execution.operator.schema.source.TimeSeriesSchemaSource.mapToString;
3437

@@ -42,8 +45,17 @@ public class TimeseriesContext {
4245

4346
private final String deadband;
4447
private final String deadbandParameters;
48+
private final int activeCountMultiplier;
49+
private final Set<String> activeLogicalViewCountSet;
4550

4651
public TimeseriesContext(IMeasurementSchemaInfo schemaInfo) {
52+
this(schemaInfo, 1, Collections.emptySet());
53+
}
54+
55+
public TimeseriesContext(
56+
IMeasurementSchemaInfo schemaInfo,
57+
int activeCountMultiplier,
58+
Set<String> activeLogicalViewCountSet) {
4759
this.dataType = schemaInfo.getSchema().getType().toString();
4860
this.encoding = schemaInfo.getSchema().getEncodingType().toString();
4961
this.compression = schemaInfo.getSchema().getCompressor().toString();
@@ -54,6 +66,8 @@ public TimeseriesContext(IMeasurementSchemaInfo schemaInfo) {
5466
MetaUtils.parseDeadbandInfo(schemaInfo.getSchema().getProps());
5567
this.deadband = deadbandInfo.left;
5668
this.deadbandParameters = deadbandInfo.right;
69+
this.activeCountMultiplier = activeCountMultiplier;
70+
this.activeLogicalViewCountSet = new HashSet<>(activeLogicalViewCountSet);
5771
}
5872

5973
public String getDataType() {
@@ -88,6 +102,14 @@ public String getDeadband() {
88102
return deadband;
89103
}
90104

105+
public int getActiveCountMultiplier() {
106+
return activeCountMultiplier;
107+
}
108+
109+
public Set<String> getActiveLogicalViewCountSet() {
110+
return activeLogicalViewCountSet;
111+
}
112+
91113
public TimeseriesContext(
92114
String dataType,
93115
String alias,
@@ -97,6 +119,30 @@ public TimeseriesContext(
97119
String attributes,
98120
String deadband,
99121
String deadbandParameters) {
122+
this(
123+
dataType,
124+
alias,
125+
encoding,
126+
compression,
127+
tags,
128+
attributes,
129+
deadband,
130+
deadbandParameters,
131+
1,
132+
Collections.emptySet());
133+
}
134+
135+
public TimeseriesContext(
136+
String dataType,
137+
String alias,
138+
String encoding,
139+
String compression,
140+
String tags,
141+
String attributes,
142+
String deadband,
143+
String deadbandParameters,
144+
int activeCountMultiplier,
145+
Set<String> activeLogicalViewCountSet) {
100146
this.dataType = dataType;
101147
this.alias = alias;
102148
this.encoding = encoding;
@@ -105,6 +151,24 @@ public TimeseriesContext(
105151
this.attributes = attributes;
106152
this.deadband = deadband;
107153
this.deadbandParameters = deadbandParameters;
154+
this.activeCountMultiplier = activeCountMultiplier;
155+
this.activeLogicalViewCountSet = new HashSet<>(activeLogicalViewCountSet);
156+
}
157+
158+
public TimeseriesContext mergeActiveCount(TimeseriesContext that) {
159+
Set<String> mergedActiveLogicalViewCountSet = new HashSet<>(activeLogicalViewCountSet);
160+
mergedActiveLogicalViewCountSet.addAll(that.activeLogicalViewCountSet);
161+
return new TimeseriesContext(
162+
dataType,
163+
alias,
164+
encoding,
165+
compression,
166+
tags,
167+
attributes,
168+
deadband,
169+
deadbandParameters,
170+
activeCountMultiplier + that.activeCountMultiplier,
171+
mergedActiveLogicalViewCountSet);
108172
}
109173

110174
public void serializeAttributes(ByteBuffer byteBuffer) {
@@ -116,6 +180,11 @@ public void serializeAttributes(ByteBuffer byteBuffer) {
116180
ReadWriteIOUtils.write(attributes, byteBuffer);
117181
ReadWriteIOUtils.write(deadband, byteBuffer);
118182
ReadWriteIOUtils.write(deadbandParameters, byteBuffer);
183+
ReadWriteIOUtils.write(activeCountMultiplier, byteBuffer);
184+
ReadWriteIOUtils.write(activeLogicalViewCountSet.size(), byteBuffer);
185+
for (String logicalView : activeLogicalViewCountSet) {
186+
ReadWriteIOUtils.write(logicalView, byteBuffer);
187+
}
119188
}
120189

121190
public void serializeAttributes(DataOutputStream stream) throws IOException {
@@ -127,6 +196,11 @@ public void serializeAttributes(DataOutputStream stream) throws IOException {
127196
ReadWriteIOUtils.write(attributes, stream);
128197
ReadWriteIOUtils.write(deadband, stream);
129198
ReadWriteIOUtils.write(deadbandParameters, stream);
199+
ReadWriteIOUtils.write(activeCountMultiplier, stream);
200+
ReadWriteIOUtils.write(activeLogicalViewCountSet.size(), stream);
201+
for (String logicalView : activeLogicalViewCountSet) {
202+
ReadWriteIOUtils.write(logicalView, stream);
203+
}
130204
}
131205

132206
public static TimeseriesContext deserialize(ByteBuffer buffer) {
@@ -138,8 +212,23 @@ public static TimeseriesContext deserialize(ByteBuffer buffer) {
138212
String attributes = ReadWriteIOUtils.readString(buffer);
139213
String deadband = ReadWriteIOUtils.readString(buffer);
140214
String deadbandParameters = ReadWriteIOUtils.readString(buffer);
215+
int activeCountMultiplier = ReadWriteIOUtils.readInt(buffer);
216+
int activeLogicalViewCountSetSize = ReadWriteIOUtils.readInt(buffer);
217+
Set<String> activeLogicalViewCountSet = new HashSet<>();
218+
for (int i = 0; i < activeLogicalViewCountSetSize; i++) {
219+
activeLogicalViewCountSet.add(ReadWriteIOUtils.readString(buffer));
220+
}
141221
return new TimeseriesContext(
142-
dataType, alias, encoding, compression, tags, attributes, deadband, deadbandParameters);
222+
dataType,
223+
alias,
224+
encoding,
225+
compression,
226+
tags,
227+
attributes,
228+
deadband,
229+
deadbandParameters,
230+
activeCountMultiplier,
231+
activeLogicalViewCountSet);
143232
}
144233

145234
@Override
@@ -159,13 +248,24 @@ public boolean equals(Object obj) {
159248
&& Objects.equals(tags, that.tags)
160249
&& Objects.equals(attributes, that.attributes)
161250
&& Objects.equals(deadband, that.deadband)
162-
&& Objects.equals(deadbandParameters, that.deadbandParameters);
251+
&& Objects.equals(deadbandParameters, that.deadbandParameters)
252+
&& activeCountMultiplier == that.activeCountMultiplier
253+
&& Objects.equals(activeLogicalViewCountSet, that.activeLogicalViewCountSet);
163254
return res;
164255
}
165256

166257
@Override
167258
public int hashCode() {
168259
return Objects.hash(
169-
dataType, alias, encoding, compression, tags, attributes, deadband, deadbandParameters);
260+
dataType,
261+
alias,
262+
encoding,
263+
compression,
264+
tags,
265+
attributes,
266+
deadband,
267+
deadbandParameters,
268+
activeCountMultiplier,
269+
activeLogicalViewCountSet);
170270
}
171271
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/CountGroupByLevelScanOperator.java‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.apache.iotdb.db.queryengine.execution.operator.schema.source.ISchemaSource;
2828
import org.apache.iotdb.db.queryengine.execution.operator.source.SourceOperator;
2929
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
30+
import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
3031
import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.ISchemaInfo;
3132
import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.reader.ISchemaReader;
3233

@@ -95,6 +96,10 @@ public OperatorContext getOperatorContext() {
9596
return operatorContext;
9697
}
9798

99+
private ISchemaRegion getSchemaRegion() {
100+
return ((SchemaDriverContext) operatorContext.getDriverContext()).getSchemaRegion();
101+
}
102+
98103
@Override
99104
public ListenableFuture<?> isBlocked() {
100105
if (isBlocked == null) {
@@ -109,6 +114,11 @@ public ListenableFuture<?> isBlocked() {
109114
*/
110115
private ListenableFuture<?> tryGetNext() {
111116
if (schemaReader == null) {
117+
if (schemaSource.shouldSkipSchemaRegion(getSchemaRegion())) {
118+
next = null;
119+
isFinished = true;
120+
return NOT_BLOCKED;
121+
}
112122
schemaReader = createTimeSeriesReader();
113123
}
114124
while (true) {
@@ -172,15 +182,14 @@ public TsBlock next() throws Exception {
172182
@Override
173183
public boolean hasNext() throws Exception {
174184
isBlocked().get(); // wait for the next TsBlock
175-
if (!schemaReader.isSuccess()) {
185+
if (schemaReader != null && !schemaReader.isSuccess()) {
176186
throw new SchemaExecutionException(schemaReader.getFailure());
177187
}
178188
return next != null;
179189
}
180190

181191
public ISchemaReader<T> createTimeSeriesReader() {
182-
return schemaSource.getSchemaReader(
183-
((SchemaDriverContext) operatorContext.getDriverContext()).getSchemaRegion());
192+
return schemaSource.getSchemaReader(getSchemaRegion());
184193
}
185194

186195
private TsBlock constructTsBlockAndClearMap(Map<PartialPath, Long> countMap) {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaCountOperator.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,10 @@ public ListenableFuture<?> isBlocked() {
9393
*/
9494
private ListenableFuture<?> tryGetNext() {
9595
ISchemaRegion schemaRegion = getSchemaRegion();
96+
if (schemaSource.shouldSkipSchemaRegion(schemaRegion)) {
97+
next = constructTsBlock(0);
98+
return NOT_BLOCKED;
99+
}
96100
if (schemaSource.hasSchemaStatistic(schemaRegion)) {
97101
next = constructTsBlock(schemaSource.getSchemaStatistic(schemaRegion));
98102
return NOT_BLOCKED;

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/ISchemaSource.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.ISchemaInfo;
2525
import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.reader.ISchemaReader;
2626

27+
import org.apache.tsfile.common.conf.TSFileDescriptor;
2728
import org.apache.tsfile.read.common.block.TsBlockBuilder;
2829

2930
import java.util.List;
@@ -52,4 +53,16 @@ public interface ISchemaSource<T extends ISchemaInfo> {
5253
boolean hasSchemaStatistic(ISchemaRegion schemaRegion);
5354

5455
long getSchemaStatistic(ISchemaRegion schemaRegion);
56+
57+
default boolean shouldSkipSchemaRegion(ISchemaRegion schemaRegion) {
58+
return false;
59+
}
60+
61+
default boolean checkRegionDatabaseIncluded(ISchemaRegion schemaRegion) {
62+
return true;
63+
}
64+
65+
default long getMaxMemory(ISchemaRegion schemaRegion) {
66+
return TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes();
67+
}
5568
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/SchemaSourceFactory.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ public static ISchemaSource<ITimeSeriesSchemaInfo> getTimeSeriesSchemaCountSourc
4444
Map<Integer, Template> templateMap,
4545
PathPatternTree scope) {
4646
return new TimeSeriesSchemaSource(
47-
pathPattern, isPrefixMatch, 0, 0, schemaFilter, templateMap, false, scope);
47+
pathPattern, isPrefixMatch, 0, 0, schemaFilter, templateMap, false, true, scope);
4848
}
4949

5050
// show time series
@@ -57,7 +57,7 @@ public static ISchemaSource<ITimeSeriesSchemaInfo> getTimeSeriesSchemaScanSource
5757
Map<Integer, Template> templateMap,
5858
PathPatternTree scope) {
5959
return new TimeSeriesSchemaSource(
60-
pathPattern, isPrefixMatch, limit, offset, schemaFilter, templateMap, true, scope);
60+
pathPattern, isPrefixMatch, limit, offset, schemaFilter, templateMap, true, false, scope);
6161
}
6262

6363
// count device

0 commit comments

Comments
 (0)