diff --git a/CHANGES.txt b/CHANGES.txt index 50235906b3..4f5bd57e7c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -81,6 +81,7 @@ * Remove compaction Severity from DynamicEndpointSnitch (CASSANDRA-11738) * Restore resumable hints delivery (CASSANDRA-11960) Merged from 3.0: + * Improve avg aggregate functions (CASSANDRA-12417) * Preserve quoted reserved keyword column names in MV creation (CASSANDRA-11803) * nodetool stopdaemon errors out (CASSANDRA-12646) * Split materialized view mutations on build to prevent OOM (CASSANDRA-12268) diff --git a/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java b/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java index 7256c6613f..4e3b97757d 100644 --- a/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java +++ b/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java @@ -19,6 +19,7 @@ package org.apache.cassandra.cql3.functions; import java.math.BigDecimal; import java.math.BigInteger; +import java.math.RoundingMode; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collection; @@ -26,6 +27,7 @@ import java.util.List; import org.apache.cassandra.cql3.CQL3Type; import org.apache.cassandra.db.marshal.*; +import org.apache.cassandra.exceptions.InvalidRequestException; /** * Factory methods for aggregate functions. @@ -102,7 +104,7 @@ public abstract class AggregateFcts public ByteBuffer compute(int protocolVersion) { - return ((LongType) returnType()).decompose(count); + return LongType.instance.decompose(count); } public void addInput(int protocolVersion, List values) @@ -149,7 +151,7 @@ public abstract class AggregateFcts if (value == null) return; - BigDecimal number = ((BigDecimal) argTypes().get(0).compose(value)); + BigDecimal number = DecimalType.instance.compose(value); sum = sum.add(number); } }; @@ -166,22 +168,19 @@ public abstract class AggregateFcts { return new Aggregate() { - private BigDecimal sum = BigDecimal.ZERO; + private BigDecimal avg = BigDecimal.ZERO; private int count; public void reset() { count = 0; - sum = BigDecimal.ZERO; + avg = BigDecimal.ZERO; } public ByteBuffer compute(int protocolVersion) { - if (count == 0) - return DecimalType.instance.decompose(BigDecimal.ZERO); - - return DecimalType.instance.decompose(sum.divide(BigDecimal.valueOf(count), BigDecimal.ROUND_HALF_EVEN)); + return DecimalType.instance.decompose(avg); } public void addInput(int protocolVersion, List values) @@ -193,7 +192,9 @@ public abstract class AggregateFcts count++; BigDecimal number = DecimalType.instance.compose(value); - sum = sum.add(number); + + // avg = avg + (value - sum) / count. + avg = avg.add(number.subtract(avg).divide(BigDecimal.valueOf(count), RoundingMode.HALF_EVEN)); } }; } @@ -229,7 +230,7 @@ public abstract class AggregateFcts if (value == null) return; - BigInteger number = ((BigInteger) argTypes().get(0).compose(value)); + BigInteger number = IntegerType.instance.compose(value); sum = sum.add(number); } }; @@ -259,9 +260,9 @@ public abstract class AggregateFcts public ByteBuffer compute(int protocolVersion) { if (count == 0) - return ((IntegerType) returnType()).decompose(BigInteger.ZERO); + return IntegerType.instance.decompose(BigInteger.ZERO); - return ((IntegerType) returnType()).decompose(sum.divide(BigInteger.valueOf(count))); + return IntegerType.instance.decompose(sum.divide(BigInteger.valueOf(count))); } public void addInput(int protocolVersion, List values) @@ -323,35 +324,11 @@ public abstract class AggregateFcts { public Aggregate newAggregate() { - return new Aggregate() + return new AvgAggregate(ByteType.instance) { - private byte sum; - - private int count; - - public void reset() + public ByteBuffer compute(int protocolVersion) throws InvalidRequestException { - count = 0; - sum = 0; - } - - public ByteBuffer compute(int protocolVersion) - { - int avg = count == 0 ? 0 : sum / count; - - return ((ByteType) returnType()).decompose((byte) avg); - } - - public void addInput(int protocolVersion, List values) - { - ByteBuffer value = values.get(0); - - if (value == null) - return; - - count++; - Number number = ((Number) argTypes().get(0).compose(value)); - sum += number.byteValue(); + return ByteType.instance.decompose((byte) computeInternal()); } }; } @@ -401,35 +378,11 @@ public abstract class AggregateFcts { public Aggregate newAggregate() { - return new Aggregate() + return new AvgAggregate(ShortType.instance) { - private short sum; - - private int count; - - public void reset() - { - count = 0; - sum = 0; - } - public ByteBuffer compute(int protocolVersion) { - int avg = count == 0 ? 0 : sum / count; - - return ((ShortType) returnType()).decompose((short) avg); - } - - public void addInput(int protocolVersion, List values) - { - ByteBuffer value = values.get(0); - - if (value == null) - return; - - count++; - Number number = ((Number) argTypes().get(0).compose(value)); - sum += number.shortValue(); + return ShortType.instance.decompose((short) computeInternal()); } }; } @@ -479,35 +432,11 @@ public abstract class AggregateFcts { public Aggregate newAggregate() { - return new Aggregate() + return new AvgAggregate(Int32Type.instance) { - private int sum; - - private int count; - - public void reset() - { - count = 0; - sum = 0; - } - public ByteBuffer compute(int protocolVersion) { - int avg = count == 0 ? 0 : sum / count; - - return ((Int32Type) returnType()).decompose(avg); - } - - public void addInput(int protocolVersion, List values) - { - ByteBuffer value = values.get(0); - - if (value == null) - return; - - count++; - Number number = ((Number) argTypes().get(0).compose(value)); - sum += number.intValue(); + return Int32Type.instance.decompose((int) computeInternal()); } }; } @@ -533,7 +462,13 @@ public abstract class AggregateFcts { public Aggregate newAggregate() { - return new LongAvgAggregate(); + return new AvgAggregate(LongType.instance) + { + public ByteBuffer compute(int protocolVersion) + { + return LongType.instance.decompose(computeInternal()); + } + }; } }; @@ -581,35 +516,11 @@ public abstract class AggregateFcts { public Aggregate newAggregate() { - return new Aggregate() + return new FloatAvgAggregate(FloatType.instance) { - private float sum; - - private int count; - - public void reset() + public ByteBuffer compute(int protocolVersion) throws InvalidRequestException { - count = 0; - sum = 0; - } - - public ByteBuffer compute(int protocolVersion) - { - float avg = count == 0 ? 0 : sum / count; - - return ((FloatType) returnType()).decompose(avg); - } - - public void addInput(int protocolVersion, List values) - { - ByteBuffer value = values.get(0); - - if (value == null) - return; - - count++; - Number number = ((Number) argTypes().get(0).compose(value)); - sum += number.floatValue(); + return FloatType.instance.decompose((float) computeInternal()); } }; } @@ -650,6 +561,95 @@ public abstract class AggregateFcts }; } }; + /** + * 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 + * converted to corresponding representation by concrete implementations. + */ + private static abstract class FloatAvgAggregate implements AggregateFunction.Aggregate + { + private double sum; + private double compensation; + private double simpleSum; + + private int count; + + private BigDecimal bigSum = null; + private boolean overflow = false; + + private final AbstractType numberType; + + public FloatAvgAggregate(AbstractType numberType) + { + this.numberType = numberType; + } + + public void reset() + { + sum = 0; + compensation = 0; + simpleSum = 0; + + count = 0; + bigSum = null; + overflow = false; + } + + public double computeInternal() + { + if (count == 0) + return 0d; + + if (overflow) + { + return bigSum.divide(BigDecimal.valueOf(count), RoundingMode.HALF_EVEN).doubleValue(); + } + else + { + // 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)) + sum = simpleSum; + else + sum = tmp; + + return sum / count; + } + } + + public void addInput(int protocolVersion, List values) + { + ByteBuffer value = values.get(0); + + if (value == null) + return; + + count++; + + double number = ((Number) numberType.compose(value)).doubleValue(); + + if (overflow) + { + bigSum = bigSum.add(BigDecimal.valueOf(number)); + } + else + { + simpleSum += number; + double prev = sum; + double tmp = number - compensation; + double rounded = sum + tmp; + compensation = (rounded - sum) - tmp; + sum = rounded; + + if (Double.isInfinite(sum) && !Double.isInfinite(number)) + { + overflow = true; + bigSum = BigDecimal.valueOf(prev).add(BigDecimal.valueOf(number)); + } + } + } + } /** * AVG function for double values. @@ -659,35 +659,11 @@ public abstract class AggregateFcts { public Aggregate newAggregate() { - return new Aggregate() + return new FloatAvgAggregate(DoubleType.instance) { - private double sum; - - private int count; - - public void reset() + public ByteBuffer compute(int protocolVersion) throws InvalidRequestException { - count = 0; - sum = 0; - } - - public ByteBuffer compute(int protocolVersion) - { - double avg = count == 0 ? 0 : sum / count; - - return ((DoubleType) returnType()).decompose(avg); - } - - public void addInput(int protocolVersion, List values) - { - ByteBuffer value = values.get(0); - - if (value == null) - return; - - count++; - Number number = ((Number) argTypes().get(0).compose(value)); - sum += number.doubleValue(); + return DoubleType.instance.decompose(computeInternal()); } }; } @@ -713,7 +689,13 @@ public abstract class AggregateFcts { public Aggregate newAggregate() { - return new LongAvgAggregate(); + return new AvgAggregate(LongType.instance) + { + public ByteBuffer compute(int protocolVersion) throws InvalidRequestException + { + return CounterColumnType.instance.decompose(computeInternal()); + } + }; } }; @@ -941,23 +923,43 @@ public abstract class AggregateFcts } } - private static class LongAvgAggregate implements AggregateFunction.Aggregate + /** + * Average aggregate class, collecting the sum using long arithmetics, falling back + * to BigInteger on long overflow. Resulting number is converted to corresponding + * representation by concrete implementations. + */ + private static abstract class AvgAggregate implements AggregateFunction.Aggregate { private long sum; - private int count; + private BigInteger bigSum = null; + private boolean overflow = false; + + private final AbstractType numberType; + + public AvgAggregate(AbstractType type) + { + this.numberType = type; + } public void reset() { count = 0; - sum = 0; + sum = 0L; + overflow = false; + bigSum = null; } - public ByteBuffer compute(int protocolVersion) + long computeInternal() { - long avg = count == 0 ? 0 : sum / count; - - return LongType.instance.decompose(avg); + if (overflow) + { + return bigSum.divide(BigInteger.valueOf(count)).longValue(); + } + else + { + return count == 0 ? 0 : (sum / count); + } } public void addInput(int protocolVersion, List values) @@ -968,8 +970,22 @@ public abstract class AggregateFcts return; count++; - Number number = LongType.instance.compose(value); - sum += number.longValue(); + long number = ((Number) numberType.compose(value)).longValue(); + if (overflow) + { + bigSum = bigSum.add(BigInteger.valueOf(number)); + } + else + { + long prev = sum; + sum += number; + + if (((prev ^ sum) & (number ^ sum)) < 0) + { + overflow = true; + bigSum = BigInteger.valueOf(prev).add(BigInteger.valueOf(number)); + } + } } } } diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java index 26f9210f8a..b01993c558 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java @@ -18,6 +18,7 @@ package org.apache.cassandra.cql3.validation.operations; import java.math.BigDecimal; +import java.math.BigInteger; import java.math.RoundingMode; import java.nio.ByteBuffer; import java.text.SimpleDateFormat; @@ -39,8 +40,6 @@ import ch.qos.logback.classic.LoggerContext; import ch.qos.logback.classic.spi.TurboFilterList; import ch.qos.logback.classic.turbo.ReconfigureOnChangeFilter; import ch.qos.logback.classic.turbo.TurboFilter; -import com.datastax.driver.core.Session; -import com.datastax.driver.core.TupleValue; import org.apache.cassandra.config.Schema; import org.apache.cassandra.config.SchemaConstants; import org.apache.cassandra.cql3.CQLTester; @@ -1969,56 +1968,110 @@ public class AggregationTest extends CQLTester } @Test - public void testSameStateInstance() throws Throwable + public void testAggregatesWithoutOverflow() throws Throwable { - // CASSANDRA-9613 removes the neccessity to re-serialize the state variable for each - // UDA state function and final function call. - // - // To test that the same state object instance is used during each invocation of the - // state and final function, this test uses a trick: - // it puts the identity hash code of the state variable to a tuple. The test then - // just asserts that the identity hash code is the same for all invocations - // of the state function and the final function. + createTable("create table %s (bucket int primary key, v1 tinyint, v2 smallint, v3 int, v4 bigint, v5 varint)"); + for (int i = 1; i <= 3; i++) + execute("insert into %s (bucket, v1, v2, v3, v4, v5) values (?, ?, ?, ?, ?, ?)", i, + (byte) ((Byte.MAX_VALUE / 3) + i), (short) ((Short.MAX_VALUE / 3) + i), (Integer.MAX_VALUE / 3) + i, (Long.MAX_VALUE / 3) + i, + BigInteger.valueOf(Long.MAX_VALUE).add(BigInteger.valueOf(i))); - String sf = createFunction(KEYSPACE, - "tuple, int", - "CREATE FUNCTION %s(s tuple, i int) " + - "CALLED ON NULL INPUT " + - "RETURNS tuple " + - "LANGUAGE java " + - "AS 's.setInt(i, System.identityHashCode(s)); return s;'"); + assertRows(execute("select avg(v1), avg(v2), avg(v3), avg(v4), avg(v5) from %s where bucket in (1, 2, 3);"), + row((byte) ((Byte.MAX_VALUE / 3) + 2), (short) ((Short.MAX_VALUE / 3) + 2), (Integer.MAX_VALUE / 3) + 2, (Long.MAX_VALUE / 3) + 2, + BigInteger.valueOf(Long.MAX_VALUE).add(BigInteger.valueOf(2)))); - String ff = createFunction(KEYSPACE, - "tuple", - "CREATE FUNCTION %s(s tuple) " + - "CALLED ON NULL INPUT " + - "RETURNS tuple " + - "LANGUAGE java " + - "AS 's.setInt(3, System.identityHashCode(s)); return s;'"); + for (int i = 1; i <= 3; i++) + execute("insert into %s (bucket, v1, v2, v3, v4, v5) values (?, ?, ?, ?, ?, ?)", i + 3, + (byte) (100 + i), (short) (100 + i), 100 + i, 100L + i, BigInteger.valueOf(100 + i)); - String a = createAggregate(KEYSPACE, - "int", - "CREATE AGGREGATE %s(int) " + - "SFUNC " + shortFunctionName(sf) + ' ' + - "STYPE tuple " + - "FINALFUNC " + shortFunctionName(ff) + ' ' + - "INITCOND (0,1,2)"); + assertRows(execute("select avg(v1), avg(v2), avg(v3), avg(v4), avg(v5) from %s where bucket in (4, 5, 6);"), + row((byte) 102, (short) 102, 102, 102L, BigInteger.valueOf(102))); + } - createTable("CREATE TABLE %s (a int primary key, b int)"); - execute("INSERT INTO %s (a, b) VALUES (0, 0)"); - execute("INSERT INTO %s (a, b) VALUES (1, 1)"); - execute("INSERT INTO %s (a, b) VALUES (2, 2)"); - try (Session s = sessionNet()) + @Test + public void testAggregateOverflow() throws Throwable + { + createTable("create table %s (bucket int primary key, v1 tinyint, v2 smallint, v3 int, v4 bigint, v5 varint)"); + for (int i = 1; i <= 3; i++) + execute("insert into %s (bucket, v1, v2, v3, v4, v5) values (?, ?, ?, ?, ?, ?)", i, + Byte.MAX_VALUE, Short.MAX_VALUE, Integer.MAX_VALUE, Long.MAX_VALUE, BigInteger.valueOf(Long.MAX_VALUE).multiply(BigInteger.valueOf(2))); + + assertRows(execute("select avg(v1), avg(v2), avg(v3), avg(v4), avg(v5) from %s where bucket in (1, 2, 3);"), + row(Byte.MAX_VALUE, Short.MAX_VALUE, Integer.MAX_VALUE, Long.MAX_VALUE, BigInteger.valueOf(Long.MAX_VALUE).multiply(BigInteger.valueOf(2)))); + + execute("truncate %s"); + + for (int i = 1; i <= 3; i++) + execute("insert into %s (bucket, v1, v2, v3, v4, v5) values (?, ?, ?, ?, ?, ?)", i, + Byte.MIN_VALUE, Short.MIN_VALUE, Integer.MIN_VALUE, Long.MIN_VALUE, BigInteger.valueOf(Long.MIN_VALUE).multiply(BigInteger.valueOf(2))); + + assertRows(execute("select avg(v1), avg(v2), avg(v3), avg(v4), avg(v5) from %s where bucket in (1, 2, 3);"), + row(Byte.MIN_VALUE, Short.MIN_VALUE, Integer.MIN_VALUE, Long.MIN_VALUE, BigInteger.valueOf(Long.MIN_VALUE).multiply(BigInteger.valueOf(2)))); + + } + + @Test + public void testDoubleAggregatesPrecision() throws Throwable + { + createTable("create table %s (bucket int primary key, v1 float, v2 double, v3 decimal)"); + + for (int i = 1; i <= 3; i++) + execute("insert into %s (bucket, v1, v2, v3) values (?, ?, ?, ?)", i, + Float.MAX_VALUE, Double.MAX_VALUE, BigDecimal.valueOf(Double.MAX_VALUE).add(BigDecimal.valueOf(2))); + + assertRows(execute("select avg(v1), avg(v2), avg(v3) from %s where bucket in (1, 2, 3);"), + row(Float.MAX_VALUE, Double.MAX_VALUE, BigDecimal.valueOf(Double.MAX_VALUE).add(BigDecimal.valueOf(2)))); + + execute("insert into %s (bucket, v1, v2, v3) values (?, ?, ?, ?)", 4, (float) 100.10, 100.10, BigDecimal.valueOf(100.10)); + execute("insert into %s (bucket, v1, v2, v3) values (?, ?, ?, ?)", 5, (float) 110.11, 110.11, BigDecimal.valueOf(110.11)); + execute("insert into %s (bucket, v1, v2, v3) values (?, ?, ?, ?)", 6, (float) 120.12, 120.12, BigDecimal.valueOf(120.12)); + + assertRows(execute("select avg(v1), avg(v2), avg(v3) from %s where bucket in (4, 5, 6);"), + row((float) 110.11, 110.11, BigDecimal.valueOf(110.11))); + } + + @Test + public void testNan() throws Throwable + { + createTable("create table %s (bucket int primary key, v1 float, v2 double)"); + + for (int i = 1; i <= 10; i++) + if (i != 5) + execute("insert into %s (bucket, v1, v2) values (?, ?, ?)", i, (float) i, (double) i); + + execute("insert into %s (bucket, v1, v2) values (?, ?, ?)", 5, Float.NaN, Double.NaN); + + 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)); + } + + @Test + public void testInfinity() throws Throwable + { + createTable("create table %s (bucket int primary key, v1 float, v2 double)"); + for (boolean positive: new boolean[] { true, false}) { - com.datastax.driver.core.Row row = s.execute("SELECT " + a + "(b) FROM " + KEYSPACE + '.' + currentTable()).one(); - TupleValue tuple = row.getTupleValue(0); - int h0 = tuple.getInt(0); - int h1 = tuple.getInt(1); - int h2 = tuple.getInt(2); - int h3 = tuple.getInt(3); - assertEquals(h0, h1); - assertEquals(h0, h2); - assertEquals(h0, h3); + final float FLOAT_INFINITY = positive ? Float.POSITIVE_INFINITY : Float.NEGATIVE_INFINITY; + final double DOUBLE_INFINITY = positive ? Double.POSITIVE_INFINITY : Double.NEGATIVE_INFINITY; + + for (int i = 1; i <= 10; i++) + if (i != 5) + execute("insert into %s (bucket, v1, v2) values (?, ?, ?)", i, (float) i, (double) i); + + execute("insert into %s (bucket, v1, v2) values (?, ?, ?)", 5, FLOAT_INFINITY, DOUBLE_INFINITY); + + 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)); + execute("truncate %s"); } } + + @Test + public void testSumPrecision() throws Throwable + { + createTable("create table %s (bucket int primary key, v1 float, v2 double, v3 decimal)"); + + 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)); + } }