You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by al...@apache.org on 2016/10/17 17:50:08 UTC
[3/8] cassandra git commit: Improve avg aggregate functions
Improve avg aggregate functions
Patch by Alex Petrov; reviewed by Branimir Lambov for CASSANDRA-12417
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/7872318d
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/7872318d
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/7872318d
Branch: refs/heads/trunk
Commit: 7872318d63009193415ba1365beedc2303a92386
Parents: 153583b
Author: Alex Petrov <ol...@gmail.com>
Authored: Wed Oct 5 09:46:05 2016 +0200
Committer: Aleksey Yeschenko <al...@apache.org>
Committed: Mon Oct 17 18:16:36 2016 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/cql3/functions/AggregateFcts.java | 327 ++++++++++---------
.../validation/operations/AggregationTest.java | 110 +++++++
3 files changed, 282 insertions(+), 156 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7872318d/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index e82eedd..d5419c6 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.10
+ * 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)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7872318d/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java b/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
index b2cae50..441fa58 100644
--- a/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
+++ b/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
@@ -27,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.
@@ -114,7 +115,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<ByteBuffer> values)
@@ -155,7 +156,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);
}
};
@@ -172,22 +173,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<ByteBuffer> values)
@@ -199,7 +197,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));
}
};
}
@@ -235,7 +235,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);
}
};
@@ -265,9 +265,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<ByteBuffer> values)
@@ -329,35 +329,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<ByteBuffer> 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());
}
};
}
@@ -407,35 +383,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<ByteBuffer> 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());
}
};
}
@@ -485,35 +437,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<ByteBuffer> 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());
}
};
}
@@ -539,7 +467,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());
+ }
+ };
}
};
@@ -587,35 +521,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<ByteBuffer> 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());
}
};
}
@@ -656,6 +566,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<ByteBuffer> 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.
@@ -665,35 +664,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()
- {
- 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<ByteBuffer> values)
+ public ByteBuffer compute(int protocolVersion) throws InvalidRequestException
{
- 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());
}
};
}
@@ -719,7 +694,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());
+ }
+ };
}
};
@@ -947,23 +928,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<ByteBuffer> values)
@@ -974,8 +975,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));
+ }
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7872318d/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
----------------------------------------------------------------------
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 411d5ee..2e7dc1a 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.cql3.validation.operations;
import java.math.BigDecimal;
import java.math.MathContext;
+import java.math.BigInteger;
import java.math.RoundingMode;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
@@ -28,6 +29,7 @@ import java.util.Date;
import java.util.Locale;
import java.util.TimeZone;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.stream.DoubleStream;
import org.apache.commons.lang3.time.DateUtils;
@@ -1969,4 +1971,112 @@ public class AggregationTest extends CQLTester
assertRows(execute("select avg(val) from %s where bucket in (1, 2, 3);"),
row(a));
}
+
+ @Test
+ public void testAggregatesWithoutOverflow() 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) ((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)));
+
+ 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))));
+
+ 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));
+
+ 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)));
+ }
+
+ @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})
+ {
+ 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));
+ }
}