Skip to content

Commit

Permalink
Optimize the serialize method of intermedia result of aggregation
Browse files Browse the repository at this point in the history
  • Loading branch information
@Beyyes
Beyyes authored Oct 23, 2024
1 parent 38f7219 commit 14346dc
Show file tree
Hide file tree
Showing 10 changed files with 256 additions and 311 deletions.
89 changes: 62 additions & 27 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 @@ -47,15 +47,15 @@ public class IoTDBMultiIDsWithAttributesTableIT {
"CREATE DATABASE db",
"USE db",
"CREATE TABLE table0 (device string id, level string id, attr1 string attribute, attr2 string attribute, num int32 measurement, bigNum int64 measurement, "
+ "floatNum double measurement, str TEXT measurement, bool BOOLEAN measurement, date DATE measurement, blob BLOB measurement, ts TIMESTAMP measurement, stringV STRING measurement)",
+ "floatNum double measurement, str TEXT measurement, bool BOOLEAN measurement, date DATE measurement, blob BLOB measurement, ts TIMESTAMP measurement, stringV STRING measurement, doubleNum DOUBLE measurement)",
"insert into table0(device, level, attr1, attr2, time,num,bigNum,floatNum,str,bool) values('d1', 'l1', 'c', 'd', 0,3,2947483648,231.2121,'coconut',FALSE)",
"insert into table0(device, level, attr1, attr2, time,num,bigNum,floatNum,str,bool,blob,ts) values('d1', 'l2', 'y', 'z', 20,2,2147483648,434.12,'pineapple',TRUE,X'108DCD62',2024-09-24T06:15:35.000+00:00)",
"insert into table0(device, level, attr1, attr2, time,num,bigNum,floatNum,str,bool,blob,ts,doubleNum) values('d1', 'l2', 'y', 'z', 20,2,2147483648,434.12,'pineapple',TRUE,X'108DCD62',2024-09-24T06:15:35.000+00:00,6666.8)",
"insert into table0(device, level, attr1, attr2, time,num,bigNum,floatNum,str,bool) values('d1', 'l3', 't', 'a', 40,1,2247483648,12.123,'apricot',TRUE)",
"insert into table0(device, level, time,num,bigNum,floatNum,str,bool) values('d1', 'l4', 80,9,2147483646,43.12,'apple',FALSE)",
"insert into table0(device, level, time,num,bigNum,floatNum,str,bool) values('d1', 'l5', 100,8,2147483964,4654.231,'papaya',TRUE)",
"insert into table0(device, level, time,num,bigNum,floatNum,str,bool) values('d1', 'l1', 31536000000,6,2147483650,1231.21,'banana',TRUE)",
"insert into table0(device, level, time,num,bigNum,floatNum,str,bool) values('d1', 'l2',31536000100,10,3147483648,231.55,'pumelo',FALSE)",
"insert into table0(device,level, time,num,bigNum,floatNum,str,bool) values('d1', 'l3',31536000500,4,2147493648,213.1,'peach',FALSE)",
"insert into table0(device,level, time,num,bigNum,floatNum,str,bool,doubleNum) values('d1', 'l3',31536000500,4,2147493648,213.1,'peach',FALSE,6666.3)",
"insert into table0(device, level, time,num,bigNum,floatNum,str,bool,blob,ts,stringV) values('d1', 'l4',31536001000,5,2149783648,56.32,'orange',FALSE,X'108DCD62',2024-09-15T06:15:35.000+00:00,'test-string1')",
"insert into table0(device, level, time,num,bigNum,floatNum,str,bool,blob,ts) values('d1', 'l5',31536010000,7,2147983648,213.112,'lemon',TRUE,X'108DCD63',2024-09-25T06:15:35.000+00:00)",
"insert into table0(device, level, time,num,bigNum,floatNum,str,bool) values('d1', 'l1',31536100000,11,2147468648,54.121,'pitaya',FALSE)",
Expand All @@ -72,7 +72,7 @@ public class IoTDBMultiIDsWithAttributesTableIT {
new String[] {
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool) values('d2','l2',20,2,2147483648,434.12,'pineapple',TRUE)",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool) values('d2','l3',40,1,2247483648,12.123,'apricot',TRUE)",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool,blob,ts,stringV) values('d2','l4',80,9,2147483646,43.12,'apple',FALSE,X'108DCD63',2024-09-20T06:15:35.000+00:00,'test-string2')",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool,blob,ts,stringV,doubleNum) values('d2','l4',80,9,2147483646,43.12,'apple',FALSE,X'108DCD63',2024-09-20T06:15:35.000+00:00,'test-string2',6666.7)",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool) values('d2','l5',100,8,2147483964,4654.231,'papaya',TRUE)",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool) values('d2','l1',31536000000,6,2147483650,1231.21,'banana',TRUE)",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool) values('d2','l3',31536000500,4,2147493648,213.1,'peach',FALSE)",
Expand All @@ -81,32 +81,43 @@ public class IoTDBMultiIDsWithAttributesTableIT {
"flush",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool,ts,StringV) values('d2','l1',31536100000,11,2147468648,54.121,'pitaya',FALSE,2024-08-01T06:15:35.000+00:00,'test-string3')",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool) values('d2','l2',41536000000,12,2146483648,45.231,'strawberry',FALSE)",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool) values('d2','l3',41536000020,14,2907483648,231.34,'cherry',FALSE)",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool,doubleNum) values('d2','l3',41536000020,14,2907483648,231.34,'cherry',FALSE,6666.4)",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool) values('d2','l4',41536900000,13,2107483648,54.12,'lychee',TRUE)",
"insert into table0(device,level,time,num,bigNum,floatNum,str,bool,date) values('d2','l5',51536000000,15,3147483648,235.213,'watermelon',TRUE,'2023-01-01')"
};

// public static void main(String[] args) {
// for (String sql : sql1) {
// System.out.println(sql+";");
// }
// for (String sql : sql2) {
// System.out.println(sql+";");
// }
// }

private static final String[] sql3 =
new String[] {
"CREATE TABLE table1 (device string id, level string id, attr1 string attribute, attr2 string attribute, num int32 measurement, bigNum int64 measurement, "
+ "floatNum double measurement, str TEXT measurement, bool BOOLEAN measurement, date DATE measurement, blob BLOB measurement)",
"insert into table1(device, level, attr1, attr2, time, num, bigNum, floatNum, str, bool, date, blob) "
+ "values('d1', 'l1', 'c', 'd', 0, 3, 2947483648,231.2121, 'coconut', FALSE, '2023-01-01', X'100DCD62')",
"insert into table1(device, level, attr1, attr2, time, num, bigNum, floatNum, str, bool, blob) "
+ "values('d1', 'l2', 'y', 'z', 20, 2, 2147483648,434.12, 'pineapple', TRUE, X'101DCD62')",
"insert into table1(device, level, attr1, attr2, time, num, bigNum, floatNum, str, bool, blob) "
+ "values('d1', 'l3', 't', 'a', 40, 1, 2247483648,12.123, 'apricot', TRUE, null)",
"insert into table1(device, level, time, num, bigNum, floatNum, str, bool, blob) "
+ "values('d1', 'l4', 80, 9, 2147483646, 43.12, 'apple', FALSE, X'104DCD62')",
"insert into table1(device, level, time, num, bigNum, floatNum, str, bool, date, blob) "
+ "values('d1', 'l5', 100, 8, 2147483964, 4654.231, 'papaya', TRUE, '2023-05-01', null)",
+ "floatNum double measurement, str TEXT measurement, bool BOOLEAN measurement, date DATE measurement, blob BLOB measurement, doubleNum DOUBLE measurement)",
"insert into table1(device, level, attr1, attr2, time, num, bigNum, floatNum, str, bool, date, blob) values('d1', 'l1', 'c', 'd', 0, 3, 2947483648,231.2121, 'coconut', FALSE, '2023-01-01', X'100DCD62')",
"insert into table1(device, level, attr1, attr2, time, num, bigNum, floatNum, str, bool, blob) values('d1', 'l2', 'y', 'z', 20, 2, 2147483648,434.12, 'pineapple', TRUE, X'101DCD62')",
"insert into table1(device, level, attr1, attr2, time, num, bigNum, floatNum, str, bool, blob) values('d1', 'l3', 't', 'a', 40, 1, 2247483648,12.123, 'apricot', TRUE, null)",
"insert into table1(device, level, time, num, bigNum, floatNum, str, bool, blob) values('d1', 'l4', 80, 9, 2147483646, 43.12, 'apple', FALSE, X'104DCD62')",
"insert into table1(device, level, time, num, bigNum, floatNum, str, bool, date, blob) values('d1', 'l5', 100, 8, 2147483964, 4654.231, 'papaya', TRUE, '2023-05-01', null)",
"flush",
"insert into table1(device, level, time, num, bigNum, floatNum, str, bool, blob) values('d1', 'l1', 31536000000, 6, 2147483650, 1231.21, 'banana', TRUE, X'106DCD62')",
"insert into table1(device, time, num, bigNum, floatNum, str, bool, blob) values('d999', 31536000000, 6, 2147483650, 1231.21, 'banana', TRUE, X'107DCD62')",
"insert into table1(level, time, num, bigNum, floatNum, str, bool, date, blob) values('l999', 31536000000, 6, 2147483650, 1231.21, 'banana', TRUE, '2023-10-01', X'108DCD62')",
"flush",
"insert into table1(time, device, level, attr1, attr2, num,bigNum,floatNum,str,bool) values(0, 'd11', 'l11', 'c', 'd', 3, 2947483648, 231.2121, 'coconut', FALSE)",
"flush",
"insert into table1(time, device, level, attr1, attr2, num,bigNum,floatNum,str,bool) values(10, 'd11', 'l11', 'c', 'd', 3, 2947483648, 231.2121, 'coconut', FALSE)",
"flush",
"insert into table1(time, device, level, attr1, attr2, num,bigNum,floatNum,str,bool) values(20, 'd11', 'l11', 'c', 'd', 3, 2947483648, 231.2121, 'coconut', FALSE)",
"flush",
"insert into table1(time, device, level, attr1, attr2, num,bigNum,floatNum,str,bool) values(30, 'd11', 'l11', 'c', 'd', 3, 2947483648, 231.2121, 'coconut', FALSE)",
"flush",
"insert into table1(device, level, time, num, bigNum, floatNum, str, bool, blob) "
+ "values('d1', 'l1', 31536000000, 6, 2147483650, 1231.21, 'banana', TRUE, X'106DCD62')",
"insert into table1(device, time, num, bigNum, floatNum, str, bool, blob) "
+ "values('d999', 31536000000, 6, 2147483650, 1231.21, 'banana', TRUE, X'107DCD62')",
"insert into table1(level, time, num, bigNum, floatNum, str, bool, date, blob) "
+ "values('l999', 31536000000, 6, 2147483650, 1231.21, 'banana', TRUE, '2023-10-01', X'108DCD62')",
"insert into table1(time, device, level, attr1, attr2, num,bigNum,floatNum,str,bool) values(40, 'd11', 'l11', 'c', 'd', 3, 2947483648, 231.2121, 'coconut', FALSE)"
};

String[] expectedHeader;
Expand Down Expand Up @@ -540,7 +551,7 @@ public void aggregationTest() {
+ "count(attr1) as count_attr1, count(device) as count_device, \n"
+ "round(avg(floatnum)) as avg_floatnum, count(date) as count_date, count(level) as count_level,"
+ "count(time) as count_time, count(*) as count_star "
+ "from table1",
+ "from table1 where device != 'd11' or device is null",
expectedHeader,
retArray,
DATABASE_NAME);
Expand Down Expand Up @@ -878,6 +889,15 @@ public void groupByDateBinTest() {
};
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);

// flush multi times, generated multi tsfile
expectedHeader = buildHeaders(1);
sql = "select date_bin(40ms,time), first(time) from table1 where device='d11' group by 1";
retArray =
new String[] {
"1970-01-01T00:00:00.000Z,1970-01-01T00:00:00.000Z,",
"1970-01-01T00:00:00.000Z,1970-01-01T00:00:00.000Z,"
};

// TODO(beyyes) test below
// sql = "select count(*) from (\n" +
// "\tselect device, level, date_bin(1d, time) as bin, \n" +
Expand Down Expand Up @@ -1008,16 +1028,22 @@ public void lastFirstMaxMinTest() {
"select first(time),first(device),first(level),first(attr1),first(attr2),first(num),first(bignum),first(floatnum),first(str),first(bool),first(date),first(blob),first(ts),first(stringv) from table0 where device='d1' and level='l4'";
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);

expectedHeader =
expectedHeader = buildHeaders(30);
retArray =
new String[] {
"_col0", "_col1", "_col2", "_col3", "_col4", "_col5",
"1971-08-20T11:33:20.000Z,d2,l5,yy,zz,15,3147483648,4654.231,watermelon,true,2023-01-01,0x108dcd63,2024-09-25T06:15:35.000Z,test-string3,6666.8,1970-01-01T00:00:00.000Z,d1,l1,c,a,1,2107483648,12.123,apple,false,2022-01-01,0x108dcd62,2024-08-01T06:15:35.000Z,test-string1,6666.3,",
};
sql =
"select max(time),max(device),max(level),max(attr1),max(attr2),max(num),max(bignum),max(floatnum),max(str),max(bool),max(date),max(blob),max(ts),max(stringv),max(doubleNum),min(time),min(device),min(level),min(attr1),min(attr2),min(num),min(bignum),min(floatnum),min(str),min(bool),min(date),min(blob),min(ts),min(stringv),min(doubleNum) from table0";
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);

expectedHeader = buildHeaders(24);
retArray =
new String[] {
"15,3147483648,4654.231,1,2107483648,12.123,",
"1971-08-20T11:33:20.000Z,d2,l5,yy,zz,15,3147483648,4654.231,2023-01-01,1971-01-01T00:01:40.000Z,test-string3,6666.8,1970-01-01T00:00:00.000Z,d1,l1,c,a,1,2107483648,12.123,2022-01-01,1970-01-01T00:00:00.020Z,test-string1,6666.3,",
};
sql =
"select max(num),max(bignum),max(floatnum),min(num),min(bignum),min(floatnum) from table0";
"select max(time),max(device),max(level),max(attr1),max(attr2),max(num),max(bignum),max(floatnum),max(date),max(ts),max(stringv),max(doubleNum),min(time),min(device),min(level),min(attr1),min(attr2),min(num),min(bignum),min(floatnum),min(date),min(ts),min(stringv),min(doubleNum) from table0";
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);

expectedHeader =
Expand Down Expand Up @@ -1137,6 +1163,15 @@ public void maxByMinByExtremeTest() {
};
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);

expectedHeader = buildHeaders(30);
sql =
"select max_by(time,blob),max_by(device,blob),max_by(level,blob),max_by(attr1,blob),max_by(attr2,blob),max_by(num,blob),max_by(bignum,blob),max_by(floatnum,blob),max_by(str,blob),max_by(bool,blob),max_by(date,blob),max_by(blob,blob),max_by(ts,blob),max_by(stringv,blob),max_by(doubleNum,blob),min_by(time,blob),min_by(device,blob),min_by(level,blob),min_by(attr1,blob),min_by(attr2,blob),min_by(num,blob),min_by(bignum,blob),min_by(floatnum,blob),min_by(str,blob),min_by(bool,blob),min_by(date,blob),min_by(blob,blob),min_by(ts,blob),min_by(stringv,blob),min_by(doubleNum,blob) from table0";
retArray =
new String[] {
"1971-01-01T00:00:10.000Z,d1,l5,null,null,7,2147983648,213.112,lemon,true,null,0x108dcd63,2024-09-25T06:15:35.000Z,null,null,1970-01-01T00:00:00.020Z,d1,l2,yy,zz,2,2147483648,434.12,pineapple,true,null,0x108dcd62,2024-09-24T06:15:35.000Z,null,6666.8,",
};
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);

expectedHeader = buildHeaders(3);
sql = "select extreme(num),extreme(bignum),extreme(floatnum) from table0";
retArray = new String[] {"15,3147483648,4654.231,"};
Expand Down
6 changes: 4 additions & 2 deletions ...b/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -459,12 +459,14 @@ private Column buildValueColumn(
private Column getIdOrAttrColumn(int positionCount, String columnName) {
if (columnName == null) {
return new RunLengthEncodedColumn(
new BinaryColumn(positionCount, Optional.of(new boolean[] {true}), new Binary[] {null}),
new BinaryColumn(1, Optional.of(new boolean[] {true}), new Binary[] {null}),
positionCount);
} else {
return new RunLengthEncodedColumn(
new BinaryColumn(
positionCount, Optional.empty(), new Binary[] {new Binary(columnName.getBytes())}),
1,
Optional.of(new boolean[] {false}),
new Binary[] {new Binary(columnName.getBytes())}),
positionCount);
}
}
Expand Down
26 changes: 6 additions & 20 deletions ...iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,10 +31,6 @@
import org.apache.tsfile.utils.RamUsageEstimator;
import org.apache.tsfile.write.UnSupportedDataTypeException;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;

import static com.google.common.base.Preconditions.checkArgument;

public class AvgAccumulator implements TableAccumulator {
Expand Down Expand Up @@ -99,15 +95,11 @@ public void addIntermediate(Column argument) {
}

initResult = true;
long midCountValue = BytesUtils.bytesToLong(argument.getBinary(i).getValues(), Long.BYTES);
double midSumValue = BytesUtils.bytesToDouble(argument.getBinary(i).getValues(), Long.BYTES);
long midCountValue = BytesUtils.bytesToLong(argument.getBinary(i).getValues(), 8);
double midSumValue = BytesUtils.bytesToDouble(argument.getBinary(i).getValues(), 8);
countValue += midCountValue;
sumValue += midSumValue;
}

if (countValue == 0) {
initResult = false;
}
}

@Override
Expand Down Expand Up @@ -161,16 +153,10 @@ public void reset() {
}

private byte[] serializeState() {
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
try {
dataOutputStream.writeLong(countValue);
dataOutputStream.writeDouble(sumValue);
} catch (IOException e) {
throw new UnsupportedOperationException(
"Failed to serialize intermediate result for AvgAccumulator.", e);
}
return byteArrayOutputStream.toByteArray();
byte[] bytes = new byte[16];
BytesUtils.longToBytes(countValue, bytes, 0);
BytesUtils.doubleToBytes(sumValue, bytes, 8);
return bytes;
}

private void addIntInput(Column column) {
Expand Down
Loading

0 comments on commit 14346dc

Please sign in to comment.