Skip to content

Commit bc31e44

Browse files
CoollZzzCoollZzz
committed
fix: address review comments for statistical aggregation functions
1 parent 39d950d commit bc31e44

14 files changed

Lines changed: 319 additions & 245 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java‎

Lines changed: 19 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -112,12 +112,12 @@ public class IoTDBTableAggregationIT {
112112
"INSERT INTO table1(time,province,city,region,device_id,color,type,s1,s4,s6,s8,s9) values (2024-09-24T06:15:55.000+00:00,'beijing','beijing','haidian','d16','yellow','BBBBBBBBBBBBBBBB',55,55.0,'beijing_haidian_yellow_B_d16_55',X'cafebabe55',2024-09-24T06:15:55.000+00:00)",
113113
// stat_table for statistical aggregation function tests (CORR, COVAR_POP, COVAR_SAMP,
114114
// REGR_SLOPE, REGR_INTERCEPT, KURTOSIS, SKEWNESS)
115-
"CREATE TABLE stat_table(device_id STRING TAG, s1 INT32 FIELD, s2 INT64 FIELD, s3 FLOAT FIELD, s4 DOUBLE FIELD, s5 BOOLEAN FIELD, s6 TEXT FIELD)",
116-
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4, s5, s6) VALUES (1, 'd1', 1, 1, 1.0, 1.0, true, 'a')",
117-
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4, s5, s6) VALUES (2, 'd1', 2, 2, 2.0, 2.0, false, 'b')",
118-
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4, s5, s6) VALUES (3, 'd1', 3, 2, 3.0, 2.0, false, 'c')",
119-
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4, s5, s6) VALUES (4, 'd1', 4, 1, 4.0, 1.0, true, 'd')",
120-
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4, s5, s6) VALUES (5, 'd1', 5, 1, 5.0, 1.0, true, 'e')",
115+
"CREATE TABLE stat_table(device_id STRING TAG, s1 INT32 FIELD, s2 INT64 FIELD, s3 FLOAT FIELD, s4 DOUBLE FIELD, s5 BOOLEAN FIELD, s6 TEXT FIELD, s7 TIMESTAMP FIELD)",
116+
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4, s5, s6, s7) VALUES (1, 'd1', 1, 1, 1.0, 1.0, true, 'a', '2026-04-13T09:15:30.000+00:00')",
117+
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4, s5, s6, s7) VALUES (2, 'd1', 2, 2, 2.0, 2.0, false, 'b', '2026-04-13T09:30:30.000+00:00')",
118+
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4, s5, s6, s7) VALUES (3, 'd1', 3, 2, 3.0, 2.0, false, 'c', '2026-04-13T09:45:30.000+00:00')",
119+
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4, s5, s6, s7) VALUES (4, 'd1', 4, 1, 4.0, 1.0, true, 'd', '2026-04-13T10:00:30.000+00:00')",
120+
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4, s5, s6, s7) VALUES (5, 'd1', 5, 1, 5.0, 1.0, true, 'e', '2026-04-13T10:15:30.000+00:00')",
121121
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4) VALUES (1, 'd2', 1, 2, 1.0, 2.0)",
122122
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4) VALUES (2, 'd2', 1, 2, 1.0, 2.0)",
123123
"INSERT INTO stat_table(time, device_id, s1, s2, s3, s4) VALUES (3, 'd2', 1, 2, 1.0, 2.0)",
@@ -5557,6 +5557,19 @@ public void statFunctionsBasicAndCrossTypeTest() {
55575557
expectedHeader,
55585558
retArray,
55595559
DATABASE_NAME);
5560+
5561+
expectedHeader = new String[] {"_col0", "_col1", "_col2", "_col3", "_col4", "_col5", "_col6"};
5562+
retArray =
5563+
new String[] {
5564+
"1.0,1800000.0,2250000.0,1.111111111111111E-6,-1973412.0333333332,0.0,-1.1999999999999993,"
5565+
};
5566+
tableResultSetEqualTest(
5567+
"select corr(s1, s7), covar_pop(s1, s7), covar_samp(s1, s7), regr_slope(s1, s7), regr_intercept(s1, s7), "
5568+
+ "skewness(s7), kurtosis(s7) "
5569+
+ "from stat_table where device_id = 'd1'",
5570+
expectedHeader,
5571+
retArray,
5572+
DATABASE_NAME);
55605573
}
55615574

55625575
@Test

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CentralMomentAccumulator.java‎

Lines changed: 33 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727

2828
public class CentralMomentAccumulator implements Accumulator {
2929

30+
private static final int INTERMEDIATE_SIZE = Long.BYTES + 4 * Double.BYTES;
31+
3032
public enum MomentType {
3133
SKEWNESS,
3234
KURTOSIS
@@ -81,20 +83,20 @@ private double getDoubleValue(Column column, int position) {
8183

8284
private void update(double value) {
8385
long n1 = count;
84-
count++;
85-
86-
double delta = value - mean;
87-
double delta_n = delta / count;
88-
double delta_n2 = delta_n * delta_n;
89-
double term1 = delta * delta_n * n1;
90-
91-
mean += delta_n;
92-
93-
m4 += term1 * delta_n2 * (count * count - 3 * count + 3) + 6 * delta_n2 * m2 - 4 * delta_n * m3;
94-
95-
m3 += term1 * delta_n * (count - 2) - 3 * delta_n * m2;
96-
97-
m2 += term1;
86+
long n = n1 + 1;
87+
double m1 = mean;
88+
double m2 = this.m2;
89+
double m3 = this.m3;
90+
double delta = value - m1;
91+
double deltaN = delta / n;
92+
double deltaN2 = deltaN * deltaN;
93+
double dm2 = delta * deltaN * n1;
94+
95+
count = n;
96+
mean = m1 + deltaN;
97+
this.m2 = m2 + dm2;
98+
this.m3 = m3 + dm2 * deltaN * (n - 2) - 3 * deltaN * m2;
99+
m4 += dm2 * deltaN2 * (n * (double) n - 3 * n + 3) + 6 * deltaN2 * m2 - 4 * deltaN * m3;
98100
}
99101

100102
@Override
@@ -125,27 +127,28 @@ private void merge(long nB, double meanB, double m2B, double m3B, double m4B) {
125127
m4 = m4B;
126128
} else {
127129
long nA = count;
128-
long nTotal = nA + nB;
129-
double delta = meanB - mean;
130+
double m1A = mean;
131+
double m2A = m2;
132+
double m3A = m3;
133+
double n = nA + nB;
134+
double delta = meanB - m1A;
130135
double delta2 = delta * delta;
131136
double delta3 = delta * delta2;
132137
double delta4 = delta2 * delta2;
133138

139+
count = (long) n;
140+
mean = (nA * m1A + nB * meanB) / n;
141+
m2 = m2A + m2B + delta2 * nA * nB / n;
142+
m3 =
143+
m3A
144+
+ m3B
145+
+ delta3 * nA * nB * (nA - nB) / (n * n)
146+
+ 3 * delta * (nA * m2B - nB * m2A) / n;
134147
m4 +=
135148
m4B
136-
+ delta4 * nA * nB * (nA * nA - nA * nB + nB * nB) / (nTotal * nTotal * nTotal)
137-
+ 6.0 * delta2 * (nA * nA * m2B + nB * nB * m2) / (nTotal * nTotal)
138-
+ 4.0 * delta * (nA * m3B - nB * m3) / nTotal;
139-
140-
m3 +=
141-
m3B
142-
+ delta3 * nA * nB * (nA - nB) / (nTotal * nTotal)
143-
+ 3.0 * delta * (nA * m2B - nB * m2) / nTotal;
144-
145-
m2 += m2B + delta2 * nA * nB / nTotal;
146-
147-
mean += delta * nB / nTotal;
148-
count = nTotal;
149+
+ delta4 * nA * nB * (nA * nA - nA * nB + nB * nB) / (n * n * n)
150+
+ 6 * delta2 * (nA * nA * m2B + nB * nB * m2A) / (n * n)
151+
+ 4 * delta * (nA * m3B - nB * m3A) / n;
149152
}
150153
}
151154

@@ -156,7 +159,7 @@ public void outputIntermediate(ColumnBuilder[] columnBuilders) {
156159
columnBuilders[0].appendNull();
157160
} else {
158161

159-
byte[] bytes = new byte[40];
162+
byte[] bytes = new byte[INTERMEDIATE_SIZE];
160163
ByteBuffer buffer = ByteBuffer.wrap(bytes);
161164
buffer.putLong(count);
162165
buffer.putDouble(mean);

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CorrelationAccumulator.java‎

Lines changed: 22 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232

3333
public class CorrelationAccumulator implements Accumulator {
3434

35+
private static final int INTERMEDIATE_SIZE = Long.BYTES + 5 * Double.BYTES;
36+
3537
public enum CorrelationType {
3638
CORR
3739
}
@@ -139,17 +141,19 @@ private void merge(
139141
m2Y = otherM2Y;
140142
c2 = otherC2;
141143
} else {
142-
long newCount = count + otherCount;
143-
double deltaX = otherMeanX - meanX;
144-
double deltaY = otherMeanY - meanY;
145-
146-
c2 += otherC2 + deltaX * deltaY * count * otherCount / newCount;
147-
m2X += otherM2X + deltaX * deltaX * count * otherCount / newCount;
148-
m2Y += otherM2Y + deltaY * deltaY * count * otherCount / newCount;
149-
150-
meanX += deltaX * otherCount / newCount;
151-
meanY += deltaY * otherCount / newCount;
152-
count = newCount;
144+
long na = count;
145+
long n = na + otherCount;
146+
double meanXValue = meanX;
147+
double meanYValue = meanY;
148+
double deltaX = otherMeanX - meanXValue;
149+
double deltaY = otherMeanY - meanYValue;
150+
151+
m2X += otherM2X + na * otherCount * deltaX * deltaX / (double) n;
152+
m2Y += otherM2Y + na * otherCount * deltaY * deltaY / (double) n;
153+
c2 += otherC2 + deltaX * deltaY * na * otherCount / (double) n;
154+
meanX = meanXValue + deltaX * otherCount / (double) n;
155+
meanY = meanYValue + deltaY * otherCount / (double) n;
156+
count = n;
153157
}
154158
}
155159

@@ -159,14 +163,15 @@ public void outputIntermediate(ColumnBuilder[] columnBuilders) {
159163
if (count == 0) {
160164
columnBuilders[0].appendNull();
161165
} else {
162-
ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES + Double.BYTES * 5);
166+
byte[] bytes = new byte[INTERMEDIATE_SIZE];
167+
ByteBuffer buffer = ByteBuffer.wrap(bytes);
163168
buffer.putLong(count);
164169
buffer.putDouble(meanX);
165170
buffer.putDouble(meanY);
166171
buffer.putDouble(m2X);
167172
buffer.putDouble(m2Y);
168173
buffer.putDouble(c2);
169-
columnBuilders[0].writeBinary(new Binary(buffer.array()));
174+
columnBuilders[0].writeBinary(new Binary(bytes));
170175
}
171176
}
172177

@@ -176,12 +181,11 @@ public void outputFinal(ColumnBuilder columnBuilder) {
176181
throw new UnsupportedOperationException("Unknown type: " + correlationType);
177182
}
178183

179-
if (count < 2) {
180-
columnBuilder.appendNull();
181-
} else if (m2X == 0 || m2Y == 0) {
182-
columnBuilder.appendNull();
184+
double result = c2 / Math.sqrt(m2X * m2Y);
185+
if (Double.isFinite(result)) {
186+
columnBuilder.writeDouble(result);
183187
} else {
184-
columnBuilder.writeDouble(c2 / Math.sqrt(m2X * m2Y));
188+
columnBuilder.appendNull();
185189
}
186190
}
187191

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CovarianceAccumulator.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232

3333
public class CovarianceAccumulator implements Accumulator {
3434

35+
private static final int INTERMEDIATE_SIZE = Long.BYTES + 3 * Double.BYTES;
36+
3537
public enum CovarianceType {
3638
COVAR_POP,
3739
COVAR_SAMP
@@ -142,12 +144,13 @@ public void outputIntermediate(ColumnBuilder[] columnBuilders) {
142144
return;
143145
}
144146

145-
ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES + Double.BYTES * 3);
147+
byte[] bytes = new byte[INTERMEDIATE_SIZE];
148+
ByteBuffer buffer = ByteBuffer.wrap(bytes);
146149
buffer.putLong(count);
147150
buffer.putDouble(meanX);
148151
buffer.putDouble(meanY);
149152
buffer.putDouble(c2);
150-
columnBuilders[0].writeBinary(new Binary(buffer.array()));
153+
columnBuilders[0].writeBinary(new Binary(bytes));
151154
}
152155

153156
@Override

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/RegressionAccumulator.java‎

Lines changed: 37 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727

2828
public class RegressionAccumulator implements Accumulator {
2929

30+
private static final int INTERMEDIATE_SIZE = Long.BYTES + 4 * Double.BYTES;
31+
3032
public enum RegressionType {
3133
REGR_SLOPE,
3234
REGR_INTERCEPT
@@ -82,17 +84,15 @@ private double getDoubleValue(Column column, int position, TSDataType dataType)
8284
}
8385

8486
private void update(double x, double y) {
85-
long newCount = count + 1;
86-
double deltaX = x - meanX;
87-
double deltaY = y - meanY;
88-
89-
meanX += deltaX / newCount;
90-
meanY += deltaY / newCount;
91-
92-
c2 += deltaX * (y - meanY);
93-
m2X += deltaX * (x - meanX);
94-
95-
count = newCount;
87+
long n = count + 1;
88+
double oldMeanX = meanX;
89+
meanX = oldMeanX + (x - oldMeanX) / n;
90+
double oldMeanY = meanY;
91+
double newMeanY = oldMeanY + (y - oldMeanY) / n;
92+
meanY = newMeanY;
93+
c2 += (x - oldMeanX) * (y - newMeanY);
94+
m2X += (x - oldMeanX) * (x - meanX);
95+
count = n;
9696
}
9797

9898
@Override
@@ -125,16 +125,19 @@ private void merge(
125125
m2X = otherM2X;
126126
c2 = otherC2;
127127
} else {
128-
long newCount = count + otherCount;
129-
double deltaX = otherMeanX - meanX;
130-
double deltaY = otherMeanY - meanY;
131-
132-
c2 += otherC2 + deltaX * deltaY * count * otherCount / newCount;
133-
m2X += otherM2X + deltaX * deltaX * count * otherCount / newCount;
134-
135-
meanX += deltaX * otherCount / newCount;
136-
meanY += deltaY * otherCount / newCount;
137-
count = newCount;
128+
long na = count;
129+
long nb = otherCount;
130+
long n = na + nb;
131+
double meanXValue = meanX;
132+
double meanYValue = meanY;
133+
double deltaX = otherMeanX - meanXValue;
134+
double deltaY = otherMeanY - meanYValue;
135+
136+
m2X += otherM2X + na * nb * deltaX * deltaX / (double) n;
137+
c2 += otherC2 + deltaX * deltaY * na * nb / (double) n;
138+
meanX = meanXValue + deltaX * nb / (double) n;
139+
meanY = meanYValue + deltaY * nb / (double) n;
140+
count = n;
138141
}
139142
}
140143

@@ -144,7 +147,7 @@ public void outputIntermediate(ColumnBuilder[] columnBuilders) {
144147
if (count == 0) {
145148
columnBuilders[0].appendNull();
146149
} else {
147-
byte[] bytes = new byte[40];
150+
byte[] bytes = new byte[INTERMEDIATE_SIZE];
148151
ByteBuffer buffer = ByteBuffer.wrap(bytes);
149152
buffer.putLong(count);
150153
buffer.putDouble(meanX);
@@ -161,20 +164,22 @@ public void outputFinal(ColumnBuilder columnBuilder) {
161164
columnBuilder.appendNull();
162165
return;
163166
}
164-
165-
if (m2X == 0) {
166-
columnBuilder.appendNull();
167-
return;
168-
}
169-
170-
double slope = c2 / m2X;
171-
172167
switch (regressionType) {
173168
case REGR_SLOPE:
174-
columnBuilder.writeDouble(slope);
169+
double slope = c2 / m2X;
170+
if (Double.isFinite(slope)) {
171+
columnBuilder.writeDouble(slope);
172+
} else {
173+
columnBuilder.appendNull();
174+
}
175175
break;
176176
case REGR_INTERCEPT:
177-
columnBuilder.writeDouble(meanY - slope * meanX);
177+
double intercept = meanY - (c2 / m2X) * meanX;
178+
if (Double.isFinite(intercept)) {
179+
columnBuilder.writeDouble(intercept);
180+
} else {
181+
columnBuilder.appendNull();
182+
}
178183
break;
179184
default:
180185
throw new UnsupportedOperationException("Unknown type: " + regressionType);

0 commit comments

Comments
 (0)