natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve sum aggregate functions 

Patch by Alex Petrov; reviewed by Branimir Lambov for CASSANDRA-12417
This commit is contained in:
Alex Petrov 2016-10-05 10:09:04 +02:00 committed by Aleksey Yeschenko
parent f5f44f69e4
commit 04cc3a9309
3 changed files with 68 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.10
* Improve sum aggregate functions (CASSANDRA-12417)
* Make cassandra.yaml docs for batch_size_*_threshold_in_kb reflect changes in CASSANDRA-10876 (CASSANDRA-12761)
* cqlsh fails to format collections when using aliases (CASSANDRA-11534)
* Check for hash conflicts in prepared statements (CASSANDRA-12733)

101
src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
View File

@ -480,29 +480,11 @@ public abstract class AggregateFcts
{
public Aggregate newAggregate()
{
return new Aggregate()
return new FloatSumAggregate(FloatType.instance)
{
private float sum;
public void reset()
public ByteBuffer compute(int protocolVersion) throws InvalidRequestException
{
sum = 0;
}
public ByteBuffer compute(int protocolVersion)
{
return ((FloatType) returnType()).decompose(sum);
}
public void addInput(int protocolVersion, List<ByteBuffer> values)
{
ByteBuffer value = values.get(0);
if (value == null)
return;
Number number = ((Number) argTypes().get(0).compose(value));
sum += number.floatValue();
return FloatType.instance.decompose((float) computeInternal());
}
};
}
@ -534,33 +516,68 @@ public abstract class AggregateFcts
{
public Aggregate newAggregate()
{
return new Aggregate()
return new FloatSumAggregate(DoubleType.instance)
{
private double sum;
public void reset()
public ByteBuffer compute(int protocolVersion) throws InvalidRequestException
{
sum = 0;
}
public ByteBuffer compute(int protocolVersion)
{
return ((DoubleType) returnType()).decompose(sum);
}
public void addInput(int protocolVersion, List<ByteBuffer> values)
{
ByteBuffer value = values.get(0);
if (value == null)
return;
Number number = ((Number) argTypes().get(0).compose(value));
sum += number.doubleValue();
return DoubleType.instance.decompose(computeInternal());
}
};
}
};
/**
* Sum aggregate function for floating point numbers, using double arithmetics and
* Kahan's algorithm to improve result precision.
*/
private static abstract class FloatSumAggregate implements AggregateFunction.Aggregate
{
private double sum;
private double compensation;
private double simpleSum;
private final AbstractType numberType;
public FloatSumAggregate(AbstractType numberType)
{
this.numberType = numberType;
}
public void reset()
{
sum = 0;
compensation = 0;
simpleSum = 0;
}
public void addInput(int protocolVersion, List<ByteBuffer> values)
{
ByteBuffer value = values.get(0);
if (value == null)
return;
double number = ((Number) numberType.compose(value)).doubleValue();
simpleSum += number;
double tmp = number - compensation;
double rounded = sum + tmp;
compensation = (rounded - sum) - tmp;
sum = rounded;
}
public double computeInternal()
{
// correctly compute final sum if it's NaN from consequently
// adding same-signed infinite values.
double tmp = sum + compensation;
if (Double.isNaN(tmp) && Double.isInfinite(simpleSum))
return simpleSum;
else
return tmp;
}
}
/**
* Average aggregate for floating point umbers, using double arithmetics and Kahan's algorithm
* to calculate sum by default, switching to BigDecimal on sum overflow. Resulting number is

8
test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
View File

@ -2043,6 +2043,8 @@ public class AggregationTest extends CQLTester
assertRows(execute("select avg(v1), avg(v2) from %s where bucket in (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);"),
row(Float.NaN, Double.NaN));
assertRows(execute("select sum(v1), sum(v2) from %s where bucket in (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);"),
row(Float.NaN, Double.NaN));
}
@Test
@ -2062,6 +2064,9 @@ public class AggregationTest extends CQLTester
assertRows(execute("select avg(v1), avg(v2) from %s where bucket in (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);"),
row(FLOAT_INFINITY, DOUBLE_INFINITY));
assertRows(execute("select sum(v1), avg(v2) from %s where bucket in (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);"),
row(FLOAT_INFINITY, DOUBLE_INFINITY));
execute("truncate %s");
}
}
@ -2073,5 +2078,8 @@ public class AggregationTest extends CQLTester
for (int i = 1; i <= 17; i++)
execute("insert into %s (bucket, v1, v2, v3) values (?, ?, ?, ?)", i, (float) (i / 10.0), i / 10.0, BigDecimal.valueOf(i / 10.0));
assertRows(execute("select sum(v1), sum(v2), sum(v3) from %s;"),
row((float) 15.3, 15.3, BigDecimal.valueOf(15.3)));
}
}