Skip to content

Commit

Permalink
Support agg(*) query in templated align by device situation
Browse files Browse the repository at this point in the history
  • Loading branch information
@Beyyes
Beyyes authored Jun 13, 2024
1 parent ece5353 commit c5d7999
Show file tree
Hide file tree
Showing 3 changed files with 182 additions and 12 deletions.
130 changes: 129 additions & 1 deletion ...ava/org/apache/iotdb/db/it/alignbydevice/IoTDBAlignByDeviceWithTemplateAggregationIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,7 @@ public void aggregationTest() {

// __endTime result is ambiguous

// not supported: group by session, condition, agg(*), agg(s1+1), count(s1+s2), non-aligned
// not supported: group by session, condition, agg(s1+1), count(s1+s2), non-aligned
// template
}

Expand Down Expand Up @@ -528,6 +528,134 @@ public void orderByTest() {
"count(s_null) in order by clause doesn't exist.");
}

@Test
public void wildCardTest() {
String[] expectedHeader =
new String[] {
"Device,max_time(s3),max_time(s1),max_time(s2),last_value(s1),last_value(s3),last_value(s1),last_value(s2)"
};
String[] retArray =
new String[] {
"root.sg1.d1,1314000000000,1314000000000,1314000000000,13.14,1314,13.14,true,",
"root.sg1.d2,1314000000001,1314000000001,1314000000001,13.15,1315,13.15,false,",
"root.sg1.d3,1314000000002,1314000000002,1314000000002,13.16,1316,13.16,false,",
"root.sg1.d4,1314000000003,1314000000003,1314000000003,13.14,1314,13.14,true,",
};
resultSetEqualTest(
"SELECT max_time(*), last_value(s1), last_value(*) FROM root.sg1.** align by device;",
expectedHeader,
retArray);
retArray =
new String[] {
"root.sg2.d1,1314000000000,1314000000000,1314000000000,13.14,1314,13.14,true,",
"root.sg2.d2,1314000000001,1314000000001,1314000000001,13.15,1315,13.15,false,",
"root.sg2.d3,1314000000002,1314000000002,1314000000002,13.16,1316,13.16,false,",
"root.sg2.d4,1314000000003,1314000000003,1314000000003,13.14,1314,13.14,true,",
};
resultSetEqualTest(
"SELECT max_time(*), last_value(s1), last_value(*) FROM root.sg2.** align by device;",
expectedHeader,
retArray);

// filter test
expectedHeader =
new String[] {"Device,max_time(s1),last_value(s3),last_value(s1),last_value(s2),count(s2)"};
retArray =
new String[] {
"root.sg1.d1,1314000000000,1314,13.14,true,3,",
"root.sg1.d2,1314000000001,1315,13.15,false,4,",
"root.sg1.d3,1314000000002,1316,13.16,false,3,",
"root.sg1.d4,1314000000003,1314,13.14,true,4,",
};
resultSetEqualTest(
"SELECT max_time(s1), last_value(*), count(s2) FROM root.sg1.** where s3>1 align by device;",
expectedHeader,
retArray);
retArray =
new String[] {
"root.sg2.d1,1314000000000,1314,13.14,true,3,",
"root.sg2.d2,1314000000001,1315,13.15,false,4,",
"root.sg2.d3,1314000000002,1316,13.16,false,3,",
"root.sg2.d4,1314000000003,1314,13.14,true,4,",
};
resultSetEqualTest(
"SELECT max_time(s1), last_value(*), count(s2) FROM root.sg2.** where s3>1 align by device;",
expectedHeader,
retArray);

// sliding window
expectedHeader = new String[] {"Time,Device,last_value(s1),last_value(s2)"};
retArray =
new String[] {
"1,root.sg1.d1,2.2,false,",
"1,root.sg1.d2,22.2,false,",
"3,root.sg1.d2,50.0,false,",
"5,root.sg1.d2,50.0,false,",
"7,root.sg1.d3,8.8,false,",
"3,root.sg1.d4,5555.5,false,",
"5,root.sg1.d4,5555.5,false,",
};
resultSetEqualTest(
"SELECT last_value(*) FROM root.sg1.** where s3+1=1316 or s2=false group by ([1,10),3ms,2ms) having avg(s1)>0 soffset 1 slimit 2 align by device;",
expectedHeader,
retArray);
retArray =
new String[] {
"1,root.sg2.d1,2.2,false,",
"1,root.sg2.d2,22.2,false,",
"3,root.sg2.d2,50.0,false,",
"5,root.sg2.d2,50.0,false,",
"7,root.sg2.d3,8.8,false,",
"3,root.sg2.d4,5555.5,false,",
"5,root.sg2.d4,5555.5,false,",
};
resultSetEqualTest(
"SELECT last_value(*) FROM root.sg2.** where s3+1=1316 or s2=false group by ([1,10),3ms,2ms) having avg(s1)>0 soffset 1 slimit 2 align by device;",
expectedHeader,
retArray);

// having
expectedHeader =
new String[] {
"Device,last_value(s3),last_value(s1),last_value(s2),first_value(s3),first_value(s1),first_value(s2)"
};
retArray =
new String[] {
"root.sg1.d2,1315,13.15,false,11,11.1,false,",
};
resultSetEqualTest(
"SELECT last_value(*), first_value(*) FROM root.sg1.** where s2=false having count(s3+s1) > 2 align by device;",
expectedHeader,
retArray);
retArray =
new String[] {
"root.sg2.d2,1315,13.15,false,11,11.1,false,",
};
resultSetEqualTest(
"SELECT last_value(*), first_value(*) FROM root.sg2.** where s2=false having count(s3+s1) > 2 align by device;",
expectedHeader,
retArray);

// not supported expression: agg1(*)+agg2(*), agg(*)/2
expectedHeader = new String[] {"Device,count(s3) + 1,count(s1) + 1,count(s2) + 1"};
retArray =
new String[] {
"root.sg1.d2,5.0,5.0,5.0,",
};
resultSetEqualTest(
"SELECT count(*)+1 FROM root.sg1.** where s2=false having count(s3+s1) > 2 align by device;",
expectedHeader,
retArray);
retArray =
new String[] {
"root.sg2.d2,5.0,5.0,5.0,",
};
resultSetEqualTest(
"SELECT count(*)+1 FROM root.sg2.** where s2=false having count(s3+s1) > 2 align by device;",
expectedHeader,
retArray);
}

protected static void insertData() {
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
Expand Down
56 changes: 45 additions & 11 deletions ...c/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAggregationAnalyze.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@
import static org.apache.iotdb.db.queryengine.plan.analyze.TemplatedAnalyze.analyzeFrom;
import static org.apache.iotdb.db.queryengine.plan.optimization.LimitOffsetPushDown.canPushDownLimitOffsetInGroupByTimeForDevice;
import static org.apache.iotdb.db.queryengine.plan.optimization.LimitOffsetPushDown.pushDownLimitOffsetInGroupByTimeForDevice;
import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME;

/** Methods in this class are used for aggregation, templated with align by device situation. */
public class TemplatedAggregationAnalyze {
Expand Down Expand Up @@ -143,24 +144,57 @@ private static boolean analyzeSelect(

Set<Expression> aggregationExpressions = new LinkedHashSet<>();
for (ResultColumn resultColumn : queryStatement.getSelectComponent().getResultColumns()) {
if (paginationController.hasCurOffset()) {
paginationController.consumeOffset();
} else if (paginationController.hasCurLimit()) {
Expression selectExpression = resultColumn.getExpression();
Expression selectExpression = resultColumn.getExpression();

if (selectExpression instanceof FunctionExpression
&& COUNT_TIME.equalsIgnoreCase(
((FunctionExpression) selectExpression).getFunctionName())) {
outputExpressions.add(new Pair<>(selectExpression, resultColumn.getAlias()));
selectExpressions.add(selectExpression);
aggregationExpressions.add(selectExpression);

analysis.getExpressionTypes().put(NodeRef.of(selectExpression), TSDataType.INT64);
((FunctionExpression) selectExpression)
.setExpressions(Collections.singletonList(new TimestampOperand()));
continue;
}

List<Expression> subExpressions;
if (selectExpression.getOutputSymbol().contains("*")) {
// when exist wildcard, only support agg(*) and count_time(*)
if (selectExpression instanceof FunctionExpression
&& "count_time"
.equalsIgnoreCase(((FunctionExpression) selectExpression).getFunctionName())) {
analysis.getExpressionTypes().put(NodeRef.of(selectExpression), TSDataType.INT64);
((FunctionExpression) selectExpression)
.setExpressions(Collections.singletonList(new TimestampOperand()));
&& selectExpression.getExpressions().size() == 1
&& "*".equalsIgnoreCase(selectExpression.getExpressions().get(0).getOutputSymbol())) {
subExpressions = new ArrayList<>();
FunctionExpression functionExpression = (FunctionExpression) selectExpression;
for (String measurement : template.getSchemaMap().keySet()) {
FunctionExpression subFunctionExpression =
new FunctionExpression(
functionExpression.getFunctionName(),
functionExpression.getFunctionAttributes(),
Collections.singletonList(
new TimeSeriesOperand(new PartialPath(new String[] {measurement}))));
subFunctionExpression.setFunctionType(functionExpression.getFunctionType());
subExpressions.add(subFunctionExpression);
}
} else {
analyzeExpressionType(analysis, selectExpression);
return false;
}
} else {
break;
subExpressions = Collections.singletonList(selectExpression);
}

for (Expression expression : subExpressions) {
if (paginationController.hasCurOffset()) {
paginationController.consumeOffset();
} else if (paginationController.hasCurLimit()) {
outputExpressions.add(new Pair<>(expression, resultColumn.getAlias()));
selectExpressions.add(expression);
aggregationExpressions.add(expression);
analyzeExpressionType(analysis, expression);
} else {
break;
}
}
}

Expand Down
8 changes: 8 additions & 0 deletions ...c/main/java/org/apache/iotdb/db/queryengine/plan/expression/multi/FunctionExpression.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,14 @@ public void setExpressions(List<Expression> expressions) {
this.expressions = expressions;
}

public FunctionType getFunctionType() {
return functionType;
}

public void setFunctionType(FunctionType functionType) {
this.functionType = functionType;
}

public String getFunctionName() {
return functionName;
}
Expand Down

0 comments on commit c5d7999

Please sign in to comment.