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

Merge branch 'cassandra-3.0' into cassandra-3.X

This commit is contained in:
Aleksey Yeschenko 2016-10-17 18:16:45 +01:00
parent e722046406 7872318d63
commit f5f44f69e4
3 changed files with 272 additions and 202 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -81,6 +81,7 @@
* Remove compaction Severity from DynamicEndpointSnitch (CASSANDRA-11738)
* Restore resumable hints delivery (CASSANDRA-11960)
Merged from 3.0:
* Improve avg aggregate functions (CASSANDRA-12417)
* Preserve quoted reserved keyword column names in MV creation (CASSANDRA-11803)
* nodetool stopdaemon errors out (CASSANDRA-12646)
* Split materialized view mutations on build to prevent OOM (CASSANDRA-12268)

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

@ -19,6 +19,7 @@ package org.apache.cassandra.cql3.functions;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.math.RoundingMode;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
@ -26,6 +27,7 @@ import java.util.List;
import org.apache.cassandra.cql3.CQL3Type;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.exceptions.InvalidRequestException;
/**
* Factory methods for aggregate functions.
@ -102,7 +104,7 @@ public abstract class AggregateFcts
public ByteBuffer compute(int protocolVersion)
{
return ((LongType) returnType()).decompose(count);
return LongType.instance.decompose(count);
}
public void addInput(int protocolVersion, List<ByteBuffer> values)
@ -149,7 +151,7 @@ public abstract class AggregateFcts
if (value == null)
return;
BigDecimal number = ((BigDecimal) argTypes().get(0).compose(value));
BigDecimal number = DecimalType.instance.compose(value);
sum = sum.add(number);
}
};
@ -166,22 +168,19 @@ public abstract class AggregateFcts
{
return new Aggregate()
{
private BigDecimal sum = BigDecimal.ZERO;
private BigDecimal avg = BigDecimal.ZERO;
private int count;
public void reset()
{
count = 0;
sum = BigDecimal.ZERO;
avg = BigDecimal.ZERO;
}
public ByteBuffer compute(int protocolVersion)
{
if (count == 0)
return DecimalType.instance.decompose(BigDecimal.ZERO);
return DecimalType.instance.decompose(sum.divide(BigDecimal.valueOf(count), BigDecimal.ROUND_HALF_EVEN));
return DecimalType.instance.decompose(avg);
}
public void addInput(int protocolVersion, List<ByteBuffer> values)
@ -193,7 +192,9 @@ public abstract class AggregateFcts
count++;
BigDecimal number = DecimalType.instance.compose(value);
sum = sum.add(number);
// avg = avg + (value - sum) / count.
avg = avg.add(number.subtract(avg).divide(BigDecimal.valueOf(count), RoundingMode.HALF_EVEN));
}
};
}
@ -229,7 +230,7 @@ public abstract class AggregateFcts
if (value == null)
return;
BigInteger number = ((BigInteger) argTypes().get(0).compose(value));
BigInteger number = IntegerType.instance.compose(value);
sum = sum.add(number);
}
};
@ -259,9 +260,9 @@ public abstract class AggregateFcts
public ByteBuffer compute(int protocolVersion)
{
if (count == 0)
return ((IntegerType) returnType()).decompose(BigInteger.ZERO);
return IntegerType.instance.decompose(BigInteger.ZERO);
return ((IntegerType) returnType()).decompose(sum.divide(BigInteger.valueOf(count)));
return IntegerType.instance.decompose(sum.divide(BigInteger.valueOf(count)));
}
public void addInput(int protocolVersion, List<ByteBuffer> values)
@ -323,35 +324,11 @@ public abstract class AggregateFcts
{
public Aggregate newAggregate()
{
return new Aggregate()
return new AvgAggregate(ByteType.instance)
{
private byte sum;
private int count;
public void reset()
public ByteBuffer compute(int protocolVersion) throws InvalidRequestException
{
count = 0;
sum = 0;
}
public ByteBuffer compute(int protocolVersion)
{
int avg = count == 0 ? 0 : sum / count;
return ((ByteType) returnType()).decompose((byte) avg);
}
public void addInput(int protocolVersion, List<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());
}
};
}
@ -401,35 +378,11 @@ public abstract class AggregateFcts
{
public Aggregate newAggregate()
{
return new Aggregate()
return new AvgAggregate(ShortType.instance)
{
private short sum;
private int count;
public void reset()
{
count = 0;
sum = 0;
}
public ByteBuffer compute(int protocolVersion)
{
int avg = count == 0 ? 0 : sum / count;
return ((ShortType) returnType()).decompose((short) avg);
}
public void addInput(int protocolVersion, List<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());
}
};
}
@ -479,35 +432,11 @@ public abstract class AggregateFcts
{
public Aggregate newAggregate()
{
return new Aggregate()
return new AvgAggregate(Int32Type.instance)
{
private int sum;
private int count;
public void reset()
{
count = 0;
sum = 0;
}
public ByteBuffer compute(int protocolVersion)
{
int avg = count == 0 ? 0 : sum / count;
return ((Int32Type) returnType()).decompose(avg);
}
public void addInput(int protocolVersion, List<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());
}
};
}
@ -533,7 +462,13 @@ public abstract class AggregateFcts
{
public Aggregate newAggregate()
{
return new LongAvgAggregate();
return new AvgAggregate(LongType.instance)
{
public ByteBuffer compute(int protocolVersion)
{
return LongType.instance.decompose(computeInternal());
}
};
}
};
@ -581,35 +516,11 @@ public abstract class AggregateFcts
{
public Aggregate newAggregate()
{
return new Aggregate()
return new FloatAvgAggregate(FloatType.instance)
{
private float sum;
private int count;
public void reset()
public ByteBuffer compute(int protocolVersion) throws InvalidRequestException
{
count = 0;
sum = 0;
}
public ByteBuffer compute(int protocolVersion)
{
float avg = count == 0 ? 0 : sum / count;
return ((FloatType) returnType()).decompose(avg);
}
public void addInput(int protocolVersion, List<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());
}
};
}
@ -650,6 +561,95 @@ public abstract class AggregateFcts
};
}
};
/**
* Average aggregate for floating point umbers, using double arithmetics and Kahan's algorithm
* to calculate sum by default, switching to BigDecimal on sum overflow. Resulting number is
* converted to corresponding representation by concrete implementations.
*/
private static abstract class FloatAvgAggregate implements AggregateFunction.Aggregate
{
private double sum;
private double compensation;
private double simpleSum;
private int count;
private BigDecimal bigSum = null;
private boolean overflow = false;
private final AbstractType numberType;
public FloatAvgAggregate(AbstractType numberType)
{
this.numberType = numberType;
}
public void reset()
{
sum = 0;
compensation = 0;
simpleSum = 0;
count = 0;
bigSum = null;
overflow = false;
}
public double computeInternal()
{
if (count == 0)
return 0d;
if (overflow)
{
return bigSum.divide(BigDecimal.valueOf(count), RoundingMode.HALF_EVEN).doubleValue();
}
else
{
// correctly compute final sum if it's NaN from consequently
// adding same-signed infinite values.
double tmp = sum + compensation;
if (Double.isNaN(tmp) && Double.isInfinite(simpleSum))
sum = simpleSum;
else
sum = tmp;
return sum / count;
}
}
public void addInput(int protocolVersion, List<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.
@ -659,35 +659,11 @@ public abstract class AggregateFcts
{
public Aggregate newAggregate()
{
return new Aggregate()
return new FloatAvgAggregate(DoubleType.instance)
{
private double sum;
private int count;
public void reset()
public ByteBuffer compute(int protocolVersion) throws InvalidRequestException
{
count = 0;
sum = 0;
}
public ByteBuffer compute(int protocolVersion)
{
double avg = count == 0 ? 0 : sum / count;
return ((DoubleType) returnType()).decompose(avg);
}
public void addInput(int protocolVersion, List<ByteBuffer> values)
{
ByteBuffer value = values.get(0);
if (value == null)
return;
count++;
Number number = ((Number) argTypes().get(0).compose(value));
sum += number.doubleValue();
return DoubleType.instance.decompose(computeInternal());
}
};
}
@ -713,7 +689,13 @@ public abstract class AggregateFcts
{
public Aggregate newAggregate()
{
return new LongAvgAggregate();
return new AvgAggregate(LongType.instance)
{
public ByteBuffer compute(int protocolVersion) throws InvalidRequestException
{
return CounterColumnType.instance.decompose(computeInternal());
}
};
}
};
@ -941,23 +923,43 @@ public abstract class AggregateFcts
}
}
private static class LongAvgAggregate implements AggregateFunction.Aggregate
/**
* Average aggregate class, collecting the sum using long arithmetics, falling back
* to BigInteger on long overflow. Resulting number is converted to corresponding
* representation by concrete implementations.
*/
private static abstract class AvgAggregate implements AggregateFunction.Aggregate
{
private long sum;
private int count;
private BigInteger bigSum = null;
private boolean overflow = false;
private final AbstractType numberType;
public AvgAggregate(AbstractType type)
{
this.numberType = type;
}
public void reset()
{
count = 0;
sum = 0;
sum = 0L;
overflow = false;
bigSum = null;
}
public ByteBuffer compute(int protocolVersion)
long computeInternal()
{
long avg = count == 0 ? 0 : sum / count;
return LongType.instance.decompose(avg);
if (overflow)
{
return bigSum.divide(BigInteger.valueOf(count)).longValue();
}
else
{
return count == 0 ? 0 : (sum / count);
}
}
public void addInput(int protocolVersion, List<ByteBuffer> values)
@ -968,8 +970,22 @@ public abstract class AggregateFcts
return;
count++;
Number number = LongType.instance.compose(value);
sum += number.longValue();
long number = ((Number) numberType.compose(value)).longValue();
if (overflow)
{
bigSum = bigSum.add(BigInteger.valueOf(number));
}
else
{
long prev = sum;
sum += number;
if (((prev ^ sum) & (number ^ sum)) < 0)
{
overflow = true;
bigSum = BigInteger.valueOf(prev).add(BigInteger.valueOf(number));
}
}
}
}
}

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

@ -18,6 +18,7 @@
package org.apache.cassandra.cql3.validation.operations;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.math.RoundingMode;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
@ -39,8 +40,6 @@ import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.spi.TurboFilterList;
import ch.qos.logback.classic.turbo.ReconfigureOnChangeFilter;
import ch.qos.logback.classic.turbo.TurboFilter;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.TupleValue;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.config.SchemaConstants;
import org.apache.cassandra.cql3.CQLTester;
@ -1969,56 +1968,110 @@ public class AggregationTest extends CQLTester
}
@Test
public void testSameStateInstance() throws Throwable
public void testAggregatesWithoutOverflow() throws Throwable
{
// CASSANDRA-9613 removes the neccessity to re-serialize the state variable for each
// UDA state function and final function call.
//
// To test that the same state object instance is used during each invocation of the
// state and final function, this test uses a trick:
// it puts the identity hash code of the state variable to a tuple. The test then
// just asserts that the identity hash code is the same for all invocations
// of the state function and the final function.
createTable("create table %s (bucket int primary key, v1 tinyint, v2 smallint, v3 int, v4 bigint, v5 varint)");
for (int i = 1; i <= 3; i++)
execute("insert into %s (bucket, v1, v2, v3, v4, v5) values (?, ?, ?, ?, ?, ?)", i,
(byte) ((Byte.MAX_VALUE / 3) + i), (short) ((Short.MAX_VALUE / 3) + i), (Integer.MAX_VALUE / 3) + i, (Long.MAX_VALUE / 3) + i,
BigInteger.valueOf(Long.MAX_VALUE).add(BigInteger.valueOf(i)));
String sf = createFunction(KEYSPACE,
"tuple<int,int,int,int>, int",
"CREATE FUNCTION %s(s tuple<int,int,int,int>, i int) " +
"CALLED ON NULL INPUT " +
"RETURNS tuple<int,int,int,int> " +
"LANGUAGE java " +
"AS 's.setInt(i, System.identityHashCode(s)); return s;'");
assertRows(execute("select avg(v1), avg(v2), avg(v3), avg(v4), avg(v5) from %s where bucket in (1, 2, 3);"),
row((byte) ((Byte.MAX_VALUE / 3) + 2), (short) ((Short.MAX_VALUE / 3) + 2), (Integer.MAX_VALUE / 3) + 2, (Long.MAX_VALUE / 3) + 2,
BigInteger.valueOf(Long.MAX_VALUE).add(BigInteger.valueOf(2))));
String ff = createFunction(KEYSPACE,
"tuple<int,int,int,int>",
"CREATE FUNCTION %s(s tuple<int,int,int,int>) " +
"CALLED ON NULL INPUT " +
"RETURNS tuple<int,int,int,int> " +
"LANGUAGE java " +
"AS 's.setInt(3, System.identityHashCode(s)); return s;'");
for (int i = 1; i <= 3; i++)
execute("insert into %s (bucket, v1, v2, v3, v4, v5) values (?, ?, ?, ?, ?, ?)", i + 3,
(byte) (100 + i), (short) (100 + i), 100 + i, 100L + i, BigInteger.valueOf(100 + i));
String a = createAggregate(KEYSPACE,
"int",
"CREATE AGGREGATE %s(int) " +
"SFUNC " + shortFunctionName(sf) + ' ' +
"STYPE tuple<int,int,int,int> " +
"FINALFUNC " + shortFunctionName(ff) + ' ' +
"INITCOND (0,1,2)");
assertRows(execute("select avg(v1), avg(v2), avg(v3), avg(v4), avg(v5) from %s where bucket in (4, 5, 6);"),
row((byte) 102, (short) 102, 102, 102L, BigInteger.valueOf(102)));
}
createTable("CREATE TABLE %s (a int primary key, b int)");
execute("INSERT INTO %s (a, b) VALUES (0, 0)");
execute("INSERT INTO %s (a, b) VALUES (1, 1)");
execute("INSERT INTO %s (a, b) VALUES (2, 2)");
try (Session s = sessionNet())
@Test
public void testAggregateOverflow() throws Throwable
{
createTable("create table %s (bucket int primary key, v1 tinyint, v2 smallint, v3 int, v4 bigint, v5 varint)");
for (int i = 1; i <= 3; i++)
execute("insert into %s (bucket, v1, v2, v3, v4, v5) values (?, ?, ?, ?, ?, ?)", i,
Byte.MAX_VALUE, Short.MAX_VALUE, Integer.MAX_VALUE, Long.MAX_VALUE, BigInteger.valueOf(Long.MAX_VALUE).multiply(BigInteger.valueOf(2)));
assertRows(execute("select avg(v1), avg(v2), avg(v3), avg(v4), avg(v5) from %s where bucket in (1, 2, 3);"),
row(Byte.MAX_VALUE, Short.MAX_VALUE, Integer.MAX_VALUE, Long.MAX_VALUE, BigInteger.valueOf(Long.MAX_VALUE).multiply(BigInteger.valueOf(2))));
execute("truncate %s");
for (int i = 1; i <= 3; i++)
execute("insert into %s (bucket, v1, v2, v3, v4, v5) values (?, ?, ?, ?, ?, ?)", i,
Byte.MIN_VALUE, Short.MIN_VALUE, Integer.MIN_VALUE, Long.MIN_VALUE, BigInteger.valueOf(Long.MIN_VALUE).multiply(BigInteger.valueOf(2)));
assertRows(execute("select avg(v1), avg(v2), avg(v3), avg(v4), avg(v5) from %s where bucket in (1, 2, 3);"),
row(Byte.MIN_VALUE, Short.MIN_VALUE, Integer.MIN_VALUE, Long.MIN_VALUE, BigInteger.valueOf(Long.MIN_VALUE).multiply(BigInteger.valueOf(2))));
}
@Test
public void testDoubleAggregatesPrecision() throws Throwable
{
createTable("create table %s (bucket int primary key, v1 float, v2 double, v3 decimal)");
for (int i = 1; i <= 3; i++)
execute("insert into %s (bucket, v1, v2, v3) values (?, ?, ?, ?)", i,
Float.MAX_VALUE, Double.MAX_VALUE, BigDecimal.valueOf(Double.MAX_VALUE).add(BigDecimal.valueOf(2)));
assertRows(execute("select avg(v1), avg(v2), avg(v3) from %s where bucket in (1, 2, 3);"),
row(Float.MAX_VALUE, Double.MAX_VALUE, BigDecimal.valueOf(Double.MAX_VALUE).add(BigDecimal.valueOf(2))));
execute("insert into %s (bucket, v1, v2, v3) values (?, ?, ?, ?)", 4, (float) 100.10, 100.10, BigDecimal.valueOf(100.10));
execute("insert into %s (bucket, v1, v2, v3) values (?, ?, ?, ?)", 5, (float) 110.11, 110.11, BigDecimal.valueOf(110.11));
execute("insert into %s (bucket, v1, v2, v3) values (?, ?, ?, ?)", 6, (float) 120.12, 120.12, BigDecimal.valueOf(120.12));
assertRows(execute("select avg(v1), avg(v2), avg(v3) from %s where bucket in (4, 5, 6);"),
row((float) 110.11, 110.11, BigDecimal.valueOf(110.11)));
}
@Test
public void testNan() throws Throwable
{
createTable("create table %s (bucket int primary key, v1 float, v2 double)");
for (int i = 1; i <= 10; i++)
if (i != 5)
execute("insert into %s (bucket, v1, v2) values (?, ?, ?)", i, (float) i, (double) i);
execute("insert into %s (bucket, v1, v2) values (?, ?, ?)", 5, Float.NaN, Double.NaN);
assertRows(execute("select avg(v1), avg(v2) from %s where bucket in (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);"),
row(Float.NaN, Double.NaN));
}
@Test
public void testInfinity() throws Throwable
{
createTable("create table %s (bucket int primary key, v1 float, v2 double)");
for (boolean positive: new boolean[] { true, false})
{
com.datastax.driver.core.Row row = s.execute("SELECT " + a + "(b) FROM " + KEYSPACE + '.' + currentTable()).one();
TupleValue tuple = row.getTupleValue(0);
int h0 = tuple.getInt(0);
int h1 = tuple.getInt(1);
int h2 = tuple.getInt(2);
int h3 = tuple.getInt(3);
assertEquals(h0, h1);
assertEquals(h0, h2);
assertEquals(h0, h3);
final float FLOAT_INFINITY = positive ? Float.POSITIVE_INFINITY : Float.NEGATIVE_INFINITY;
final double DOUBLE_INFINITY = positive ? Double.POSITIVE_INFINITY : Double.NEGATIVE_INFINITY;
for (int i = 1; i <= 10; i++)
if (i != 5)
execute("insert into %s (bucket, v1, v2) values (?, ?, ?)", i, (float) i, (double) i);
execute("insert into %s (bucket, v1, v2) values (?, ?, ?)", 5, FLOAT_INFINITY, DOUBLE_INFINITY);
assertRows(execute("select avg(v1), avg(v2) from %s where bucket in (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);"),
row(FLOAT_INFINITY, DOUBLE_INFINITY));
execute("truncate %s");
}
}
@Test
public void testSumPrecision() throws Throwable
{
createTable("create table %s (bucket int primary key, v1 float, v2 double, v3 decimal)");
for (int i = 1; i <= 17; i++)
execute("insert into %s (bucket, v1, v2, v3) values (?, ?, ?, ?)", i, (float) (i / 10.0), i / 10.0, BigDecimal.valueOf(i / 10.0));
}
}