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:06 UTC
[1/8] cassandra git commit: Improve avg aggregate functions
Repository: cassandra
Updated Branches:
refs/heads/cassandra-3.0 153583be5 -> 7872318d6
refs/heads/cassandra-3.X e72204640 -> 04cc3a930
refs/heads/trunk b3a0130d1 -> 4cd774747
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/cassandra-3.0
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));
+ }
}
[6/8] cassandra git commit: Improve sum aggregate functions
Posted by al...@apache.org.
Improve sum 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/04cc3a93
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/04cc3a93
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/04cc3a93
Branch: refs/heads/trunk
Commit: 04cc3a9309fdc4a8c9ae33ed00d2b681a6bb117a
Parents: f5f44f6
Author: Alex Petrov <ol...@gmail.com>
Authored: Wed Oct 5 10:09:04 2016 +0200
Committer: Aleksey Yeschenko <al...@apache.org>
Committed: Mon Oct 17 18:28:18 2016 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/cql3/functions/AggregateFcts.java | 99 ++++++++++++--------
.../validation/operations/AggregationTest.java | 8 ++
3 files changed, 67 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/04cc3a93/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4f5bd57..d230462 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -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)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/04cc3a93/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 4e3b977..530b7ba 100644
--- a/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
+++ b/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
@@ -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()
- {
- sum = 0;
- }
-
- public ByteBuffer compute(int protocolVersion)
- {
- return ((FloatType) returnType()).decompose(sum);
- }
-
- public void addInput(int protocolVersion, List<ByteBuffer> values)
+ public ByteBuffer compute(int protocolVersion) throws InvalidRequestException
{
- 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;
+ return DoubleType.instance.decompose(computeInternal());
}
+ };
+ }
+ };
- public ByteBuffer compute(int protocolVersion)
- {
- return ((DoubleType) returnType()).decompose(sum);
- }
+ /**
+ * 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;
- public void addInput(int protocolVersion, List<ByteBuffer> values)
- {
- ByteBuffer value = values.get(0);
+ private final AbstractType numberType;
- if (value == null)
- return;
+ 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;
+ }
+ }
- Number number = ((Number) argTypes().get(0).compose(value));
- sum += number.doubleValue();
- }
- };
- }
- };
/**
* 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
http://git-wip-us.apache.org/repos/asf/cassandra/blob/04cc3a93/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 b01993c..8f03635 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
@@ -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)));
}
}
[2/8] cassandra git commit: Improve avg aggregate functions
Posted by al...@apache.org.
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/cassandra-3.X
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));
+ }
}
[8/8] cassandra git commit: Merge branch 'cassandra-3.X' into trunk
Posted by al...@apache.org.
Merge branch 'cassandra-3.X' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/4cd77474
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/4cd77474
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/4cd77474
Branch: refs/heads/trunk
Commit: 4cd774747e83c3e0443b66a4df674ef651c22b81
Parents: b3a0130 04cc3a9
Author: Aleksey Yeschenko <al...@apache.org>
Authored: Mon Oct 17 18:28:34 2016 +0100
Committer: Aleksey Yeschenko <al...@apache.org>
Committed: Mon Oct 17 18:28:34 2016 +0100
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../cassandra/cql3/functions/AggregateFcts.java | 425 ++++++++++---------
.../validation/operations/AggregationTest.java | 155 +++++--
3 files changed, 339 insertions(+), 243 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/4cd77474/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 63ed4a6,d230462..97bdd3b
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,11 -1,5 +1,12 @@@
+4.0
+ * Add column definition kind to dropped columns in schema (CASSANDRA-12705)
+ * Add (automate) Nodetool Documentation (CASSANDRA-12672)
+ * Update bundled cqlsh python driver to 3.7.0 (CASSANDRA-12736)
+ * Reject invalid replication settings when creating or altering a keyspace (CASSANDRA-12681)
+
+
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)
[5/8] cassandra git commit: Merge branch 'cassandra-3.0' into
cassandra-3.X
Posted by al...@apache.org.
Merge branch 'cassandra-3.0' into cassandra-3.X
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f5f44f69
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f5f44f69
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f5f44f69
Branch: refs/heads/trunk
Commit: f5f44f69e46fec7e2eb7242dcfb2484baa631371
Parents: e722046 7872318
Author: Aleksey Yeschenko <al...@apache.org>
Authored: Mon Oct 17 18:16:45 2016 +0100
Committer: Aleksey Yeschenko <al...@apache.org>
Committed: Mon Oct 17 18:19:30 2016 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/cql3/functions/AggregateFcts.java | 328 ++++++++++---------
.../validation/operations/AggregationTest.java | 147 ++++++---
3 files changed, 273 insertions(+), 203 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f5f44f69/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 5023590,d5419c6..4f5bd57
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,86 -1,5 +1,87 @@@
-3.0.10
+3.10
+ * 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)
+ * Exit query parsing upon first error (CASSANDRA-12598)
+ * Fix cassandra-stress to use single seed in UUID generation (CASSANDRA-12729)
+ * CQLSSTableWriter does not allow Update statement (CASSANDRA-12450)
+ * Config class uses boxed types but DD exposes primitive types (CASSANDRA-12199)
+ * Add pre- and post-shutdown hooks to Storage Service (CASSANDRA-12461)
+ * Add hint delivery metrics (CASSANDRA-12693)
+ * Remove IndexInfo cache from FileIndexInfoRetriever (CASSANDRA-12731)
+ * ColumnIndex does not reuse buffer (CASSANDRA-12502)
+ * cdc column addition still breaks schema migration tasks (CASSANDRA-12697)
+ * Upgrade metrics-reporter dependencies (CASSANDRA-12089)
+ * Tune compaction thread count via nodetool (CASSANDRA-12248)
+ * Add +=/-= shortcut syntax for update queries (CASSANDRA-12232)
+ * Include repair session IDs in repair start message (CASSANDRA-12532)
+ * Add a blocking task to Index, run before joining the ring (CASSANDRA-12039)
+ * Fix NPE when using CQLSSTableWriter (CASSANDRA-12667)
+ * Support optional backpressure strategies at the coordinator (CASSANDRA-9318)
+ * Make randompartitioner work with new vnode allocation (CASSANDRA-12647)
+ * Fix cassandra-stress graphing (CASSANDRA-12237)
+ * Allow filtering on partition key columns for queries without secondary indexes (CASSANDRA-11031)
+ * Fix Cassandra Stress reporting thread model and precision (CASSANDRA-12585)
+ * Add JMH benchmarks.jar (CASSANDRA-12586)
+ * Add row offset support to SASI (CASSANDRA-11990)
+ * Cleanup uses of AlterTableStatementColumn (CASSANDRA-12567)
+ * Add keep-alive to streaming (CASSANDRA-11841)
+ * Tracing payload is passed through newSession(..) (CASSANDRA-11706)
+ * avoid deleting non existing sstable files and improve related log messages (CASSANDRA-12261)
+ * json/yaml output format for nodetool compactionhistory (CASSANDRA-12486)
+ * Retry all internode messages once after a connection is
+ closed and reopened (CASSANDRA-12192)
+ * Add support to rebuild from targeted replica (CASSANDRA-9875)
+ * Add sequence distribution type to cassandra stress (CASSANDRA-12490)
+ * "SELECT * FROM foo LIMIT ;" does not error out (CASSANDRA-12154)
+ * Define executeLocally() at the ReadQuery Level (CASSANDRA-12474)
+ * Extend read/write failure messages with a map of replica addresses
+ to error codes in the v5 native protocol (CASSANDRA-12311)
+ * Fix rebuild of SASI indexes with existing index files (CASSANDRA-12374)
+ * Let DatabaseDescriptor not implicitly startup services (CASSANDRA-9054, 12550)
+ * Fix clustering indexes in presence of static columns in SASI (CASSANDRA-12378)
+ * Fix queries on columns with reversed type on SASI indexes (CASSANDRA-12223)
+ * Added slow query log (CASSANDRA-12403)
+ * Count full coordinated request against timeout (CASSANDRA-12256)
+ * Allow TTL with null value on insert and update (CASSANDRA-12216)
+ * Make decommission operation resumable (CASSANDRA-12008)
+ * Add support to one-way targeted repair (CASSANDRA-9876)
+ * Remove clientutil jar (CASSANDRA-11635)
+ * Fix compaction throughput throttle (CASSANDRA-12366, CASSANDRA-12717)
+ * Delay releasing Memtable memory on flush until PostFlush has finished running (CASSANDRA-12358)
+ * Cassandra stress should dump all setting on startup (CASSANDRA-11914)
+ * Make it possible to compact a given token range (CASSANDRA-10643)
+ * Allow updating DynamicEndpointSnitch properties via JMX (CASSANDRA-12179)
+ * Collect metrics on queries by consistency level (CASSANDRA-7384)
+ * Add support for GROUP BY to SELECT statement (CASSANDRA-10707)
+ * Deprecate memtable_cleanup_threshold and update default for memtable_flush_writers (CASSANDRA-12228)
+ * Upgrade to OHC 0.4.4 (CASSANDRA-12133)
+ * Add version command to cassandra-stress (CASSANDRA-12258)
+ * Create compaction-stress tool (CASSANDRA-11844)
+ * Garbage-collecting compaction operation and schema option (CASSANDRA-7019)
+ * Add beta protocol flag for v5 native protocol (CASSANDRA-12142)
+ * Support filtering on non-PRIMARY KEY columns in the CREATE
+ MATERIALIZED VIEW statement's WHERE clause (CASSANDRA-10368)
+ * Unify STDOUT and SYSTEMLOG logback format (CASSANDRA-12004)
+ * COPY FROM should raise error for non-existing input files (CASSANDRA-12174)
+ * Faster write path (CASSANDRA-12269)
+ * Option to leave omitted columns in INSERT JSON unset (CASSANDRA-11424)
+ * Support json/yaml output in nodetool tpstats (CASSANDRA-12035)
+ * Expose metrics for successful/failed authentication attempts (CASSANDRA-10635)
+ * Prepend snapshot name with "truncated" or "dropped" when a snapshot
+ is taken before truncating or dropping a table (CASSANDRA-12178)
+ * Optimize RestrictionSet (CASSANDRA-12153)
+ * cqlsh does not automatically downgrade CQL version (CASSANDRA-12150)
+ * Omit (de)serialization of state variable in UDAs (CASSANDRA-9613)
+ * Create a system table to expose prepared statements (CASSANDRA-8831)
+ * Reuse DataOutputBuffer from ColumnIndex (CASSANDRA-11970)
+ * Remove DatabaseDescriptor dependency from SegmentedFile (CASSANDRA-11580)
+ * Add supplied username to authentication error messages (CASSANDRA-12076)
+ * Remove pre-startup check for open JMX port (CASSANDRA-12074)
+ * 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)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f5f44f69/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f5f44f69/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
index 26f9210,2e7dc1a..b01993c
--- 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,8 +18,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;
@@@ -39,10 -42,7 +40,8 @@@ import ch.qos.logback.classic.LoggerCon
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;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
[7/8] cassandra git commit: Improve sum aggregate functions
Posted by al...@apache.org.
Improve sum 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/04cc3a93
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/04cc3a93
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/04cc3a93
Branch: refs/heads/cassandra-3.X
Commit: 04cc3a9309fdc4a8c9ae33ed00d2b681a6bb117a
Parents: f5f44f6
Author: Alex Petrov <ol...@gmail.com>
Authored: Wed Oct 5 10:09:04 2016 +0200
Committer: Aleksey Yeschenko <al...@apache.org>
Committed: Mon Oct 17 18:28:18 2016 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/cql3/functions/AggregateFcts.java | 99 ++++++++++++--------
.../validation/operations/AggregationTest.java | 8 ++
3 files changed, 67 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/04cc3a93/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4f5bd57..d230462 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -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)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/04cc3a93/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 4e3b977..530b7ba 100644
--- a/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
+++ b/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
@@ -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()
- {
- sum = 0;
- }
-
- public ByteBuffer compute(int protocolVersion)
- {
- return ((FloatType) returnType()).decompose(sum);
- }
-
- public void addInput(int protocolVersion, List<ByteBuffer> values)
+ public ByteBuffer compute(int protocolVersion) throws InvalidRequestException
{
- 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;
+ return DoubleType.instance.decompose(computeInternal());
}
+ };
+ }
+ };
- public ByteBuffer compute(int protocolVersion)
- {
- return ((DoubleType) returnType()).decompose(sum);
- }
+ /**
+ * 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;
- public void addInput(int protocolVersion, List<ByteBuffer> values)
- {
- ByteBuffer value = values.get(0);
+ private final AbstractType numberType;
- if (value == null)
- return;
+ 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;
+ }
+ }
- Number number = ((Number) argTypes().get(0).compose(value));
- sum += number.doubleValue();
- }
- };
- }
- };
/**
* 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
http://git-wip-us.apache.org/repos/asf/cassandra/blob/04cc3a93/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 b01993c..8f03635 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
@@ -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)));
}
}
[4/8] cassandra git commit: Merge branch 'cassandra-3.0' into
cassandra-3.X
Posted by al...@apache.org.
Merge branch 'cassandra-3.0' into cassandra-3.X
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f5f44f69
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f5f44f69
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f5f44f69
Branch: refs/heads/cassandra-3.X
Commit: f5f44f69e46fec7e2eb7242dcfb2484baa631371
Parents: e722046 7872318
Author: Aleksey Yeschenko <al...@apache.org>
Authored: Mon Oct 17 18:16:45 2016 +0100
Committer: Aleksey Yeschenko <al...@apache.org>
Committed: Mon Oct 17 18:19:30 2016 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/cql3/functions/AggregateFcts.java | 328 ++++++++++---------
.../validation/operations/AggregationTest.java | 147 ++++++---
3 files changed, 273 insertions(+), 203 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f5f44f69/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 5023590,d5419c6..4f5bd57
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,86 -1,5 +1,87 @@@
-3.0.10
+3.10
+ * 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)
+ * Exit query parsing upon first error (CASSANDRA-12598)
+ * Fix cassandra-stress to use single seed in UUID generation (CASSANDRA-12729)
+ * CQLSSTableWriter does not allow Update statement (CASSANDRA-12450)
+ * Config class uses boxed types but DD exposes primitive types (CASSANDRA-12199)
+ * Add pre- and post-shutdown hooks to Storage Service (CASSANDRA-12461)
+ * Add hint delivery metrics (CASSANDRA-12693)
+ * Remove IndexInfo cache from FileIndexInfoRetriever (CASSANDRA-12731)
+ * ColumnIndex does not reuse buffer (CASSANDRA-12502)
+ * cdc column addition still breaks schema migration tasks (CASSANDRA-12697)
+ * Upgrade metrics-reporter dependencies (CASSANDRA-12089)
+ * Tune compaction thread count via nodetool (CASSANDRA-12248)
+ * Add +=/-= shortcut syntax for update queries (CASSANDRA-12232)
+ * Include repair session IDs in repair start message (CASSANDRA-12532)
+ * Add a blocking task to Index, run before joining the ring (CASSANDRA-12039)
+ * Fix NPE when using CQLSSTableWriter (CASSANDRA-12667)
+ * Support optional backpressure strategies at the coordinator (CASSANDRA-9318)
+ * Make randompartitioner work with new vnode allocation (CASSANDRA-12647)
+ * Fix cassandra-stress graphing (CASSANDRA-12237)
+ * Allow filtering on partition key columns for queries without secondary indexes (CASSANDRA-11031)
+ * Fix Cassandra Stress reporting thread model and precision (CASSANDRA-12585)
+ * Add JMH benchmarks.jar (CASSANDRA-12586)
+ * Add row offset support to SASI (CASSANDRA-11990)
+ * Cleanup uses of AlterTableStatementColumn (CASSANDRA-12567)
+ * Add keep-alive to streaming (CASSANDRA-11841)
+ * Tracing payload is passed through newSession(..) (CASSANDRA-11706)
+ * avoid deleting non existing sstable files and improve related log messages (CASSANDRA-12261)
+ * json/yaml output format for nodetool compactionhistory (CASSANDRA-12486)
+ * Retry all internode messages once after a connection is
+ closed and reopened (CASSANDRA-12192)
+ * Add support to rebuild from targeted replica (CASSANDRA-9875)
+ * Add sequence distribution type to cassandra stress (CASSANDRA-12490)
+ * "SELECT * FROM foo LIMIT ;" does not error out (CASSANDRA-12154)
+ * Define executeLocally() at the ReadQuery Level (CASSANDRA-12474)
+ * Extend read/write failure messages with a map of replica addresses
+ to error codes in the v5 native protocol (CASSANDRA-12311)
+ * Fix rebuild of SASI indexes with existing index files (CASSANDRA-12374)
+ * Let DatabaseDescriptor not implicitly startup services (CASSANDRA-9054, 12550)
+ * Fix clustering indexes in presence of static columns in SASI (CASSANDRA-12378)
+ * Fix queries on columns with reversed type on SASI indexes (CASSANDRA-12223)
+ * Added slow query log (CASSANDRA-12403)
+ * Count full coordinated request against timeout (CASSANDRA-12256)
+ * Allow TTL with null value on insert and update (CASSANDRA-12216)
+ * Make decommission operation resumable (CASSANDRA-12008)
+ * Add support to one-way targeted repair (CASSANDRA-9876)
+ * Remove clientutil jar (CASSANDRA-11635)
+ * Fix compaction throughput throttle (CASSANDRA-12366, CASSANDRA-12717)
+ * Delay releasing Memtable memory on flush until PostFlush has finished running (CASSANDRA-12358)
+ * Cassandra stress should dump all setting on startup (CASSANDRA-11914)
+ * Make it possible to compact a given token range (CASSANDRA-10643)
+ * Allow updating DynamicEndpointSnitch properties via JMX (CASSANDRA-12179)
+ * Collect metrics on queries by consistency level (CASSANDRA-7384)
+ * Add support for GROUP BY to SELECT statement (CASSANDRA-10707)
+ * Deprecate memtable_cleanup_threshold and update default for memtable_flush_writers (CASSANDRA-12228)
+ * Upgrade to OHC 0.4.4 (CASSANDRA-12133)
+ * Add version command to cassandra-stress (CASSANDRA-12258)
+ * Create compaction-stress tool (CASSANDRA-11844)
+ * Garbage-collecting compaction operation and schema option (CASSANDRA-7019)
+ * Add beta protocol flag for v5 native protocol (CASSANDRA-12142)
+ * Support filtering on non-PRIMARY KEY columns in the CREATE
+ MATERIALIZED VIEW statement's WHERE clause (CASSANDRA-10368)
+ * Unify STDOUT and SYSTEMLOG logback format (CASSANDRA-12004)
+ * COPY FROM should raise error for non-existing input files (CASSANDRA-12174)
+ * Faster write path (CASSANDRA-12269)
+ * Option to leave omitted columns in INSERT JSON unset (CASSANDRA-11424)
+ * Support json/yaml output in nodetool tpstats (CASSANDRA-12035)
+ * Expose metrics for successful/failed authentication attempts (CASSANDRA-10635)
+ * Prepend snapshot name with "truncated" or "dropped" when a snapshot
+ is taken before truncating or dropping a table (CASSANDRA-12178)
+ * Optimize RestrictionSet (CASSANDRA-12153)
+ * cqlsh does not automatically downgrade CQL version (CASSANDRA-12150)
+ * Omit (de)serialization of state variable in UDAs (CASSANDRA-9613)
+ * Create a system table to expose prepared statements (CASSANDRA-8831)
+ * Reuse DataOutputBuffer from ColumnIndex (CASSANDRA-11970)
+ * Remove DatabaseDescriptor dependency from SegmentedFile (CASSANDRA-11580)
+ * Add supplied username to authentication error messages (CASSANDRA-12076)
+ * Remove pre-startup check for open JMX port (CASSANDRA-12074)
+ * 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)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f5f44f69/src/java/org/apache/cassandra/cql3/functions/AggregateFcts.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f5f44f69/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java
index 26f9210,2e7dc1a..b01993c
--- 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,8 +18,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;
@@@ -39,10 -42,7 +40,8 @@@ import ch.qos.logback.classic.LoggerCon
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;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
[3/8] cassandra git commit: Improve avg aggregate functions
Posted by al...@apache.org.
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));
+ }
}