Skip to content

Commit

Permalink
Add last_by and first_by aggregation support for table model
Browse files Browse the repository at this point in the history
  • Loading branch information
@Beyyes
Beyyes authored Oct 18, 2024
1 parent 9ea87df commit 48593d7
Show file tree
Hide file tree
Showing 17 changed files with 1,208 additions and 82 deletions.
69 changes: 69 additions & 0 deletions ...rc/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -989,6 +989,67 @@ public void lastFirstMaxMinTest() {
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
}

@Test
public void lastByFirstByTest() {
String[] expectedHeader1 = buildHeaders(13);
String[] expectedHeader2 = buildHeaders(14);

sql =
"select last_by(time,time),last_by(device,time),last_by(level,time),last_by(attr1,time),last_by(attr2,time),last_by(num,time),last_by(bignum,time),last_by(floatnum,time),last_by(str,time),last_by(bool,time),last_by(date,time),last_by(ts,time),last_by(stringv,time) from table0 where device='d2'";
retArray =
new String[] {
"1971-08-20T11:33:20.000Z,d2,l5,null,null,15,3147483648,235.213,watermelon,true,2023-01-01,null,null,",
};
tableResultSetEqualTest(sql, expectedHeader1, retArray, DATABASE_NAME);
sql =
"select last_by(time,time),last_by(device,time),last_by(level,time),last_by(attr1,time),last_by(attr2,time),last_by(num,time),last_by(bignum,time),last_by(floatnum,time),last_by(str,time),last_by(bool,time),last_by(date,time),last_by(ts,time),last_by(stringv,time),last_by(blob,time) from table0 where device='d2'";
retArray =
new String[] {
"1971-08-20T11:33:20.000Z,d2,l5,null,null,15,3147483648,235.213,watermelon,true,2023-01-01,null,null,null,",
};
tableResultSetEqualTest(sql, expectedHeader2, retArray, DATABASE_NAME);

sql =
"select last_by(time,time),last_by(time,device),last_by(time,level),last_by(time,attr1),last_by(time,attr2),last_by(time,num),last_by(time,bignum),last_by(time,floatnum),last_by(time,str),last_by(time,bool),last_by(time,date),last_by(time,ts),last_by(time,stringv) from table0 where device='d2'";
retArray =
new String[] {
"1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-04-26T17:46:40.000Z,1971-01-01T00:01:40.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-01-01T00:01:40.000Z,1971-01-01T00:01:40.000Z,",
};
tableResultSetEqualTest(sql, expectedHeader1, retArray, DATABASE_NAME);
sql =
"select last_by(time,time),last_by(time,device),last_by(time,level),last_by(time,attr1),last_by(time,attr2),last_by(time,num),last_by(time,bignum),last_by(time,floatnum),last_by(time,str),last_by(time,bool),last_by(time,date),last_by(time,ts),last_by(time,stringv),last_by(time,blob) from table0 where device='d2'";
retArray =
new String[] {
"1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-04-26T17:46:40.000Z,1971-01-01T00:01:40.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-01-01T00:01:40.000Z,1971-01-01T00:01:40.000Z,1970-01-01T00:00:00.080Z,",
};
tableResultSetEqualTest(sql, expectedHeader2, retArray, DATABASE_NAME);

String[] expectedHeader11 = buildHeaders(expectedHeader1.length * 2);
sql =
"select last_by(time,time),last_by(device,time),last_by(level,time),last_by(attr1,time),last_by(attr2,time),last_by(num,time),last_by(bignum,time),last_by(floatnum,time),last_by(str,time),last_by(bool,time),last_by(date,time),last_by(ts,time),last_by(stringv,time),last_by(time,time),last_by(time,device),last_by(time,level),last_by(time,attr1),last_by(time,attr2),last_by(time,num),last_by(time,bignum),last_by(time,floatnum),last_by(time,str),last_by(time,bool),last_by(time,date),last_by(time,ts),last_by(time,stringv) from table0 where device='d2'";
retArray =
new String[] {
"1971-08-20T11:33:20.000Z,d2,l5,null,null,15,3147483648,235.213,watermelon,true,2023-01-01,null,null,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-04-26T17:46:40.000Z,1971-01-01T00:01:40.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-01-01T00:01:40.000Z,1971-01-01T00:01:40.000Z,",
};
tableResultSetEqualTest(sql, expectedHeader11, retArray, DATABASE_NAME);

sql =
"select first_by(time,time),first_by(device,time),first_by(level,time),first_by(attr1,time),first_by(attr2,time),first_by(num,time),first_by(bignum,time),first_by(floatnum,time),first_by(str,time),first_by(bool,time),first_by(date,time),first_by(ts,time),first_by(stringv,time) from table0 where device='d2' and time>80";
retArray =
new String[] {
"1970-01-01T00:00:00.100Z,d2,l5,null,null,8,2147483964,4654.231,papaya,true,null,null,null,",
};
tableResultSetEqualTest(sql, expectedHeader1, retArray, DATABASE_NAME);

sql =
"select first_by(time,time),first_by(time,device),first_by(time,level),first_by(time,attr1),first_by(time,attr2),first_by(time,num),first_by(time,bignum),first_by(time,floatnum),first_by(time,str),first_by(time,bool),first_by(time,date),first_by(time,ts),first_by(time,stringv) from table0 where device='d2' and time>80";
retArray =
new String[] {
"1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1971-01-01T00:00:00.000Z,1971-01-01T00:00:00.000Z,1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1971-08-20T11:33:20.000Z,1971-01-01T00:01:40.000Z,1971-01-01T00:01:40.000Z,",
};
tableResultSetEqualTest(sql, expectedHeader1, retArray, DATABASE_NAME);
}

// ==================================================================
// ============================ Join Test ===========================
// ==================================================================
Expand Down Expand Up @@ -1274,4 +1335,12 @@ public void fullOuterJoinTest2() {
+ "ORDER BY time, t1.device, t2.device";
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
}

public static String[] buildHeaders(int length) {
String[] expectedHeader = new String[length];
for (int i = 0; i < length; i++) {
expectedHeader[i] = "_col" + i;
}
return expectedHeader;
}
}
62 changes: 29 additions & 33 deletions ...b/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,17 +23,18 @@
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedAccumulator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedAvgAccumulator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedCountAccumulator;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
import org.apache.iotdb.db.queryengine.plan.expression.binary.CompareBinaryExpression;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.ConstantOperand;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SymbolReference;

import org.apache.tsfile.enums.TSDataType;

import java.util.List;
import java.util.Map;

import static com.google.common.base.Preconditions.checkState;
import static org.apache.iotdb.db.queryengine.plan.relational.metadata.TableBuiltinAggregationFunction.LAST;
import static org.apache.iotdb.commons.schema.table.TsTable.TIME_COLUMN_NAME;
import static org.apache.iotdb.db.queryengine.plan.relational.metadata.TableBuiltinAggregationFunction.FIRST_BY;
import static org.apache.iotdb.db.queryengine.plan.relational.metadata.TableBuiltinAggregationFunction.LAST_BY;

public class AccumulatorFactory {

Expand All @@ -47,6 +48,23 @@ public static TableAccumulator createAccumulator(
if (aggregationType == TAggregationType.UDAF) {
// If UDAF accumulator receives raw input, it needs to check input's attribute
throw new UnsupportedOperationException();
} else if ((LAST_BY.getFunctionName().equals(functionName)
|| FIRST_BY.getFunctionName().equals(functionName))
&& inputExpressions.size() > 1) {
boolean xIsTimeColumn = false;
boolean yIsTimeColumn = false;
if (isTimeColumn(inputExpressions.get(1))) {
yIsTimeColumn = true;
} else if (isTimeColumn(inputExpressions.get(0))) {
xIsTimeColumn = true;
}
if (LAST_BY.getFunctionName().equals(functionName)) {
return new LastByAccumulator(
inputDataTypes.get(0), inputDataTypes.get(1), xIsTimeColumn, yIsTimeColumn);
} else {
return new FirstByAccumulator(
inputDataTypes.get(0), inputDataTypes.get(1), xIsTimeColumn, yIsTimeColumn);
}
} else {
return createBuiltinAccumulator(
aggregationType, inputDataTypes, inputExpressions, inputAttributes, ascending);
Expand Down Expand Up @@ -106,6 +124,10 @@ public static TableAccumulator createBuiltinAccumulator(
return new MaxAccumulator(inputDataTypes.get(0));
case MIN:
return new MinAccumulator(inputDataTypes.get(0));
case LAST_BY:
return new LastByAccumulator(inputDataTypes.get(0), inputDataTypes.get(1), false, false);
case FIRST_BY:
return new FirstByAccumulator(inputDataTypes.get(0), inputDataTypes.get(1), false, false);
default:
throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
}
Expand Down Expand Up @@ -219,34 +241,8 @@ public interface KeepEvaluator {
boolean apply(long keep);
}

public static KeepEvaluator initKeepEvaluator(Expression keepExpression) {
// We have checked semantic in FE,
// keep expression must be ConstantOperand or CompareBinaryExpression here
if (keepExpression instanceof ConstantOperand) {
return keep -> keep >= Long.parseLong(keepExpression.getExpressionString());
} else {
long constant =
Long.parseLong(
((CompareBinaryExpression) keepExpression)
.getRightExpression()
.getExpressionString());
switch (keepExpression.getExpressionType()) {
case LESS_THAN:
return keep -> keep < constant;
case LESS_EQUAL:
return keep -> keep <= constant;
case GREATER_THAN:
return keep -> keep > constant;
case GREATER_EQUAL:
return keep -> keep >= constant;
case EQUAL_TO:
return keep -> keep == constant;
case NON_EQUAL:
return keep -> keep != constant;
default:
throw new IllegalArgumentException(
"unsupported expression type: " + keepExpression.getExpressionType());
}
}
public static boolean isTimeColumn(Expression expression) {
return expression instanceof SymbolReference
&& TIME_COLUMN_NAME.equals(((SymbolReference) expression).getName());
}
}
2 changes: 1 addition & 1 deletion ...iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ public boolean hasFinalResult() {

@Override
public void addStatistics(Statistics[] statistics) {
if (statistics == null) {
if (statistics == null || statistics[0] == null) {
return;
}
initResult = true;
Expand Down
3 changes: 2 additions & 1 deletion ...tdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,9 +82,10 @@ public boolean hasFinalResult() {

@Override
public void addStatistics(Statistics[] statistics) {
if (statistics[0] == null) {
if (statistics == null || statistics[0] == null) {
return;
}

countState += statistics[0].getCount();
}

Expand Down
4 changes: 2 additions & 2 deletions ...tdb/db/queryengine/execution/operator/source/relational/aggregation/FirstAccumulator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -196,12 +196,12 @@ public void evaluateFinal(ColumnBuilder columnBuilder) {

@Override
public boolean hasFinalResult() {
return initResult;
return false;
}

@Override
public void addStatistics(Statistics[] statistics) {
if (statistics[0] == null) {
if (statistics == null || statistics[0] == null) {
return;
}
switch (seriesDataType) {
Expand Down
Loading

0 comments on commit 48593d7

Please sign in to comment.